Repository: diennea/herddb (java)

Issue:
BRIN empty search
BlockRangeIndexConcurrentTest fails with next exception

`java.lang.IllegalStateException: Empty Search! i 664
  at herddb.index.brin.BlockRangeIndexConcurrentTest.lambda$testConcurrentReadsWritesWithSplits$1(BlockRangeIndexConcurrentTest.java:149)
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
  at java.util.concurrent.FutureTask.run(FutureTask.java:266)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  at java.lang.Thread.run(Thread.java:748)`

Pull request:
BRIN keep next block reference fixes #215
BRIN blocks now keep track of next even after checkpoint, deletion and reboot.

Patch:
diff --git a/herddb-core/src/main/java/herddb/index/brin/BRINIndexManager.java b/herddb-core/src/main/java/herddb/index/brin/BRINIndexManager.java
index c87f68e2..f947e16a 100644
--- a/herddb-core/src/main/java/herddb/index/brin/BRINIndexManager.java
+++ b/herddb-core/src/main/java/herddb/index/brin/BRINIndexManager.java
@@ -46,6 +46,7 @@ import herddb.index.IndexOperation;
 import herddb.index.SecondaryIndexPrefixScan;
 import herddb.index.SecondaryIndexRangeScan;
 import herddb.index.SecondaryIndexSeek;
+import herddb.index.brin.BlockRangeIndexMetadata.BlockMetadata;
 import herddb.log.CommitLog;
 import herddb.log.LogSequenceNumber;
 import herddb.model.Index;
@@ -112,10 +113,30 @@ public class BRINIndexManager extends AbstractIndexManager {
                     out.writeVInt(metadata.size());
                     for (BlockRangeIndexMetadata.BlockMetadata<Bytes> md : metadata) {
                         /* First key is null if is the head block */
-                        out.writeArray(md.firstKey == null ? null : md.firstKey.data);
-                        out.writeVInt(md.blockId);
+                        boolean head = md.firstKey == null;
+                        /* Next block is null if is the tail block */
+                        boolean tail = md.nextBlockId == null;
+
+                        /* Prepares hasd/tail flags */
+                        byte blockFlags = 0;
+                        if (head) {
+                            blockFlags |= BlockMetadata.HEAD;
+                        }
+                        if (tail) {
+                            blockFlags |= BlockMetadata.TAIL;
+                        }
+                        out.writeByte(blockFlags);
+
+                        if (!head) {
+                            out.writeArray(md.firstKey.data);
+                        }
+                        out.writeVLong(md.blockId);
                         out.writeVLong(md.size);
                         out.writeVLong(md.pageId);
+
+                        if (!tail) {
+                            out.writeVLong(md.nextBlockId);
+                        }
                     }
                     break;
                 case TYPE_BLOCKDATA:
@@ -150,14 +171,31 @@ public class BRINIndexManager extends AbstractIndexManager {
                     int blocks = in.readVInt();
                     result.metadata = new ArrayList<>();
                     for (int i = 0; i < blocks; i++) {
-                        /* First key is null if is the head block */
-                        byte[] fk = in.readArray();
-                        Bytes firstKey = (fk == null) ? null : Bytes.from_array(fk);
-                        int blockId = in.readVInt();
+                        byte blockFlags = in.readByte();
+                        Bytes firstKey;
+                        if ((blockFlags & BlockMetadata.HEAD) > 0) {
+                            /* Is HEAD */
+                            /* First key is null if is the head block */
+                            firstKey = null;
+                        } else {
+                            firstKey = Bytes.from_array(in.readArray());
+                        }
+
+                        long blockId = in.readVLong();
                         long size = in.readVLong();
                         long pageId = in.readVLong();
+
+                        Long nextBlockId;
+                        if ((blockFlags & BlockMetadata.TAIL) > 0) {
+                            /* Is TAIL */
+                            /* Next block is null if is the tail block */
+                            nextBlockId = null;
+                        } else {
+                            nextBlockId = in.readVLong();
+                        }
+
                         BlockRangeIndexMetadata.BlockMetadata<Bytes> md
-                            = new BlockRangeIndexMetadata.BlockMetadata<>(firstKey, blockId, size, pageId);
+                            = new BlockRangeIndexMetadata.BlockMetadata<>(firstKey, blockId, size, pageId, nextBlockId);
                         result.metadata.add(md);
                     }
                     break;
diff --git a/herddb-core/src/main/java/herddb/index/brin/BlockRangeIndex.java b/herddb-core/src/main/java/herddb/index/brin/BlockRangeIndex.java
index f578f3a5..17dda043 100644
--- a/herddb-core/src/main/java/herddb/index/brin/BlockRangeIndex.java
+++ b/herddb-core/src/main/java/herddb/index/brin/BlockRangeIndex.java
@@ -33,7 +33,7 @@ import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentNavigableMap;
 import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -43,7 +43,8 @@ import java.util.stream.Stream;
 import herddb.core.Page;
 import herddb.core.Page.Metadata;
 import herddb.core.PageReplacementPolicy;
-import herddb.utils.EnsureIntegerIncrementAccumulator;
+import herddb.storage.DataStorageManagerException;
+import herddb.utils.EnsureLongIncrementAccumulator;
 import herddb.utils.SizeAwareObject;
 
 /**
@@ -61,7 +62,7 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
 
     private final long maxPageBlockSize;
     private final ConcurrentNavigableMap<BlockStartKey<K>, Block<K,V>> blocks = new ConcurrentSkipListMap<>();
-    private final AtomicInteger blockIdGenerator = new AtomicInteger();
+    private final AtomicLong currentBlockId = new AtomicLong(0L);
 
     private final IndexDataStorage<K, V> dataStorage;
 
@@ -83,10 +84,10 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
 
     static final class BlockStartKey<K extends Comparable<K>> implements Comparable<BlockStartKey<K>> {
 
-        static final BlockStartKey<?> HEAD_KEY = new BlockStartKey<>(null, -1);
+        static final BlockStartKey<?> HEAD_KEY = new BlockStartKey<>(null, 0L);
 
         public final K minKey;
-        public final int blockId;
+        public final long blockId;
 
         @Override
         public String toString() {
@@ -97,9 +98,9 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
             }
         }
 
-        public static final <X extends Comparable<X>> BlockStartKey<X> valueOf(X minKey, int segmentId) {
+        public static final <X extends Comparable<X>> BlockStartKey<X> valueOf(X minKey, long segmentId) {
             if (minKey == null) {
-                if (segmentId != -1) {
+                if (segmentId != HEAD_KEY.blockId) {
                     throw new IllegalArgumentException();
                 }
                 return (BlockStartKey<X>) HEAD_KEY;
@@ -107,7 +108,7 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
             return new BlockStartKey<>(minKey, segmentId);
         }
 
-        private BlockStartKey(K minKey, int segmentId) {
+        private BlockStartKey(K minKey, long segmentId) {
             this.minKey = minKey;
             this.blockId = segmentId;
         }
@@ -125,7 +126,7 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
             if (diff != 0) {
                 return diff;
             }
-            return Integer.compare(blockId, o.blockId);
+            return Long.compare(blockId, o.blockId);
         }
 
         public int compareMinKey(K other) {
@@ -142,7 +143,7 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
         public int hashCode() {
             int hash = 3;
             hash = 67 * hash + Objects.hashCode(this.minKey);
-            hash = 67 * hash + this.blockId;
+            hash = 67 * hash + Long.hashCode(this.blockId);
             return hash;
         }
 
@@ -218,13 +219,15 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
         private volatile boolean dirty;
         private volatile long pageId;
 
-        private final BRINPage page;
+        private final BRINPage<KEY,VAL> page;
 
-        public Block(BlockRangeIndex<KEY,VAL> index, int blockId, KEY firstKey, long size, long pageId) {
+        public Block(BlockRangeIndex<KEY,VAL> index, long blockId, KEY firstKey, long size, long pageId, Block<KEY,VAL> next) {
             this.index = index;
             this.key = BlockStartKey.valueOf(firstKey, blockId);
 
             this.size = size;
+            this.next = next;
+
             this.loaded = false;
             this.dirty = false;
             this.pageId = pageId;
@@ -233,6 +236,7 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
             page = new BRINPage<>(this, blockId);
         }
 
+        @Deprecated
         public Block(BlockRangeIndex<KEY,VAL> index, KEY firstKey, VAL firstValue) {
             this.index = index;
             this.key = (BlockStartKey<KEY>) BlockStartKey.HEAD_KEY;
@@ -243,6 +247,7 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
             values = new TreeMap<>();
             values.put(firstKey, firstKeyValues);
             this.size = index.evaluateEntrySize(firstKey, firstValue);
+
             this.loaded = true;
             this.dirty = true;
             this.pageId = IndexDataStorage.NEW_PAGE;
@@ -251,11 +256,15 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
             page = new BRINPage<>(this, key.blockId);
         }
 
-        private Block(BlockRangeIndex<KEY,VAL> index, KEY newOtherMinKey, NavigableMap<KEY, List<VAL>> other_values, long size) {
+        /** Construtor for split operations */
+        private Block(BlockRangeIndex<KEY,VAL> index, KEY minKey, NavigableMap<KEY, List<VAL>> values, long size, Block<KEY,VAL> next) {
             this.index = index;
-            this.key = BlockStartKey.valueOf(newOtherMinKey, index.blockIdGenerator.incrementAndGet());
-            this.values = other_values;
+            this.key = BlockStartKey.valueOf(minKey, index.currentBlockId.incrementAndGet());
+
+            this.values = values;
             this.size = size;
+            this.next = next;
+
             this.loaded = true;
             this.dirty = true;
             this.pageId = IndexDataStorage.NEW_PAGE;
@@ -285,20 +294,22 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
             valuesForKey.add(value);
         }
 
-        boolean addValue(KEY key, VAL value, ConcurrentNavigableMap<BlockStartKey<KEY>, Block<KEY,VAL>> blocks) {
+        Block<KEY,VAL> addValue(KEY key, VAL value) {
 
             /* Eventual new block from split. It must added to PageReplacementPolicy only after lock release */
             Block<KEY,VAL> newblock = null;
             lock.lock();
             try {
-                Block<KEY,VAL> _next = next;
-                if (_next != null && _next.key.compareMinKey(key) <= 0) {
+                Block<KEY,VAL> next = this.next;
+                if (next != null && next.key.compareMinKey(key) <= 0) {
                     // unfortunately this occours during split
                     // put #1 -> causes split
                     // put #2 -> designates this block for put, but the split is taking place
                     // put #1 returns
                     // put #2 needs to addValue to the 'next' (split result) block not to this
-                    return false;
+
+                    /* Add to next */
+                    return next;
                 }
                 ensureBlockLoaded();
 
@@ -307,7 +318,7 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
                 dirty = true;
 
                 if (size > index.maxPageBlockSize) {
-                    newblock = split(blocks);
+                    newblock = split();
                 }
             } finally {
                 lock.unlock();
@@ -319,7 +330,9 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
                     unload.owner.unload(unload.pageId);
                 }
             }
-            return true;
+
+            /* Added */
+            return null;
         }
 
 
@@ -506,7 +519,7 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
         /**
          * Return the newly generated block if any
          */
-        private Block<KEY,VAL> split(ConcurrentNavigableMap<BlockStartKey<KEY>, Block<KEY,VAL>> blocks) {
+        private Block<KEY,VAL> split() {
             if (size < index.maxPageBlockSize) {
                 throw new IllegalStateException("Split on a non overflowing block");
             }
@@ -514,8 +527,8 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
             if (LOG.isLoggable(Level.FINE)) {
                 LOG.log(Level.FINE, "Split: FK " + key, new Object[]{key});
             }
-            NavigableMap<KEY, List<VAL>> keep_values = new TreeMap<>();
-            NavigableMap<KEY, List<VAL>> other_values = new TreeMap<>();
+            NavigableMap<KEY, List<VAL>> keepValues = new TreeMap<>();
+            NavigableMap<KEY, List<VAL>> otherValues = new TreeMap<>();
 
             final long splitSize = size / 2;
 
@@ -526,26 +539,27 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
                 for (VAL v : entry.getValue()) {
                     final long entrySize = index.evaluateEntrySize(key, v);
                     if (mySize < splitSize) {
-                        mergeAddValue(key, v, keep_values);
+                        mergeAddValue(key, v, keepValues);
                         mySize += entrySize;
                     } else {
-                        mergeAddValue(key, v, other_values);
+                        mergeAddValue(key, v, otherValues);
                         otherSize += entrySize;
                     }
                 }
             }
 
-            if (other_values.isEmpty()) {
+            if (otherValues.isEmpty()) {
                 return null;
             }
 
-            KEY newOtherMinKey = other_values.firstKey();
+            KEY newOtherMinKey = otherValues.firstKey();
 
-            Block<KEY,VAL> newblock = new Block<>(index, newOtherMinKey, other_values, otherSize);
+            Block<KEY,VAL> newblock = new Block<>(index, newOtherMinKey, otherValues, otherSize, next);
 
             this.next = newblock;
             this.size = mySize;
-            this.values = keep_values;
+            this.values = keepValues;
+
 
             /*
              * First publish the new block then reduce this block min/max keys. If done otherwise a
@@ -555,7 +569,7 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
              * blocked [happen before])
              */
             // access to external field, this is the cause of most of the concurrency problems
-            blocks.put(newblock.key, newblock);
+            index.blocks.put(newblock.key, newblock);
 
             return newblock;
 
@@ -565,7 +579,8 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
             lock.lock();
             try {
                 if (!dirty || !loaded) {
-                    return new BlockRangeIndexMetadata.BlockMetadata<>(key.minKey, key.blockId, size, pageId);
+                    final Long nextBlockId = next == null ? null : next.key.blockId;
+                    return new BlockRangeIndexMetadata.BlockMetadata<>(key.minKey, key.blockId, size, pageId, nextBlockId);
                 }
                 List<Map.Entry<KEY,VAL>> result = new ArrayList<>();
                 values.forEach((k, l) -> {
@@ -581,14 +596,15 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
                 this.dirty = false;
                 this.pageId = newPageId;
 
-                return new BlockRangeIndexMetadata.BlockMetadata<>(key.minKey, key.blockId, size, pageId);
+                final Long nextBlockId = next == null ? null : next.key.blockId;
+                return new BlockRangeIndexMetadata.BlockMetadata<>(key.minKey, key.blockId, size, pageId, nextBlockId);
             } finally {
                 lock.unlock();
             }
 
         }
 
-        private boolean unload() throws IOException {
+        boolean unload() throws IOException {
             lock.lock();
             try {
                 if (!loaded) {
@@ -639,7 +655,7 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
 
         @Override
         public String toString() {
-            return "Block{" + "key=" + key + ", minKey=" + key.minKey + ", size=" + size;
+            return "Block{" + "key=" + key + ", minKey=" + key.minKey + ", size=" + size + ", next=" + (next == null ? null : next.key) + "}";
         }
 
     }
@@ -649,7 +665,7 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
     }
 
     public void put(K key, V value) {
-        BlockStartKey<K> lookUp = BlockStartKey.valueOf(key, Integer.MAX_VALUE);
+        BlockStartKey<K> lookUp = BlockStartKey.valueOf(key, Long.MAX_VALUE);
         while (!tryPut(key, value, lookUp)) {
         }
     }
@@ -658,18 +674,56 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
         final boolean fineEnabled = LOG.isLoggable(Level.FINE);
 
         List<BlockRangeIndexMetadata.BlockMetadata<K>> blocksMetadata = new ArrayList<>();
-        Set<Block<K,V>> deleted = new HashSet<>();
 
-        Iterator<Block<K,V>> iterator = blocks.values().iterator();
+        /* Reverse ordering! */
+        Iterator<Block<K,V>> iterator = blocks.descendingMap().values().iterator();
+        Block<K,V> nextReference = null;
+        boolean deletionStream = false;
+
         while(iterator.hasNext()) {
             Block<K,V> block = iterator.next();
 
             BlockRangeIndexMetadata.BlockMetadata<K> metadata = block.checkpoint();
             /* Do not discard head block */
             if (metadata.size != 0 || block.key == BlockStartKey.HEAD_KEY) {
+
+                /*
+                 * If this is the first non empty block after a deletion "stream" end the stream
+                 * and change next block to real not empty next block
+                 */
+                if (deletionStream) {
+                    deletionStream = false;
+
+                    if (fineEnabled) {
+                        if (block.next != null) {
+                            LOG.fine("unlinking block " + block.pageId + " (" + block.key + ") from deleted block "
+                                    + block.next.pageId + " (" + block.next.key + ")");
+                        }
+                        if (nextReference != null) {
+                            LOG.fine("linking block " + block.pageId + " (" + block.key + ") to real not empty next block "
+                                    + nextReference.next.pageId + " (" + nextReference.next.key + ")");
+                        }
+                    }
+
+                    /*
+                     * Changes next reference, done with lock to force next publishing.
+                     *
+                     * TODO: it could be done better.
+                     */
+                    block.lock.lock();
+                    try {
+                        block.next = nextReference;
+                        metadata.nextBlockId = nextReference == null ? null : nextReference.key.blockId;
+                    } finally {
+                        block.lock.unlock();
+                    }
+
+                }
+
                 if (fineEnabled) {
                     LOG.fine("block " + block.pageId + " ("+ block.key+ ") has " + metadata.size + " records at checkpoint");
                 }
+
                 blocksMetadata.add(metadata);
             } else {
                 LOG.info("block " + block.pageId + " ("+ block.key+ ") is empty at checkpoint: discarding");
@@ -680,26 +734,24 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
                 /* Remove the block from knowledge */
                 iterator.remove();
 
-                deleted.add(block);
-            }
-        }
-
-        if (!deleted.isEmpty()) {
-            /* Loop again to remove next pointers to delete blocks */
-            for (Block<K,V> block : blocks.values()) {
-                final Block<K,V> next = block.next;
-                if (deleted.contains(next)) {
-                    if (fineEnabled) {
-                        LOG.warning("unlinking block " + block.pageId + " ("+ block.key+ ") from deleted block " + next.pageId + " ("+ block.key+ ")");
-                    }
-                    block.next = null;
+                /*
+                 * If not already in a deletion stream save the next not empty block reference,
+                 * it will be needed at the end of block deletion stream. Otherwise current
+                 * block next block is already deleted too, we must not save his reference!
+                 */
+                if (!deletionStream) {
+                    /* Save deleted next reference */
+                    nextReference = block.next;
                 }
+
+                deletionStream = true;
             }
         }
 
         return new BlockRangeIndexMetadata<>(blocksMetadata);
     }
 
+
     private boolean tryPut(K key, V value, BlockStartKey<K> lookUp) {
         Map.Entry<BlockStartKey<K>, Block<K,V>> segmentEntry = blocks.floorEntry(lookUp);
         if (segmentEntry == null) {
@@ -723,8 +775,15 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
                 headBlock.lock.unlock();
             }
         }
-        Block<K,V> choosenSegment = segmentEntry.getValue();
-        return choosenSegment.addValue(key, value, blocks);
+
+
+        Block<K,V> block = segmentEntry.getValue();
+
+        do {
+            block = block.addValue(key, value);
+        } while(block != null);
+
+        return true;
 
     }
 
@@ -768,11 +827,11 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
         BlockStartKey<K> ceil = null;
 
         if (firstKey!= null) {
-            floor = blocks.floorKey(BlockStartKey.valueOf(firstKey, -1));
+            floor = blocks.floorKey(BlockStartKey.valueOf(firstKey, -1L));
         }
 
         if (lastKey != null) {
-            ceil = blocks.ceilingKey(BlockStartKey.valueOf(lastKey, Integer.MAX_VALUE));
+            ceil = blocks.ceilingKey(BlockStartKey.valueOf(lastKey, Long.MAX_VALUE));
         }
 
         if (floor == null) {
@@ -820,22 +879,38 @@ public final class BlockRangeIndex<K extends Comparable<K> & SizeAwareObject, V
         return !lookUpRange(key, key).isEmpty();
     }
 
-    public void boot(BlockRangeIndexMetadata<K> metadata) {
+    public void boot(BlockRangeIndexMetadata<K> metadata) throws DataStorageManagerException {
         LOG.severe("boot index, with " + metadata.getBlocksMetadata().size() + " blocks");
 
         clear();
 
+        /* Metadata are saved/recovered in reverse order so "next" block has been already created */
+        Block<K,V> next = null;
         for (BlockRangeIndexMetadata.BlockMetadata<K> blockData : metadata.getBlocksMetadata()) {
             /*
              * TODO: if the system is restart with a different (smaller) page size old blocks will remain
              * bigger until a split occurs.
              */
-            Block<K,V> block = new Block<>(this, blockData.blockId, blockData.firstKey, blockData.size, blockData.pageId);
-            blocks.put(block.key, block);
+
+            /* Medatada safety check (do not trust blindly ordering) */
+            if (blockData.nextBlockId != null) {
+                if (next == null || next.key.blockId != blockData.nextBlockId.longValue()) {
+                    throw new DataStorageManagerException("Wron next block, expected " + next.key.blockId + " but "
+                            + blockData.nextBlockId + " found");
+                }
+            } else {
+                if (next != null) {
+                    throw new DataStorageManagerException("Wron next block, expected notingh but "
+                            + blockData.nextBlockId + " found");
+                }
+            }
+
+            next = new Block<>(this, blockData.blockId, blockData.firstKey, blockData.size, blockData.pageId, next);
+            blocks.put(next.key, next);
             if (LOG.isLoggable(Level.FINE)) {
-                LOG.fine("boot block at " + block.key + " " + block.key.minKey);
+                LOG.fine("boot block at " + next.key + " " + next.key.minKey);
             }
-            blockIdGenerator.accumulateAndGet(block.key.blockId, EnsureIntegerIncrementAccumulator.INSTANCE);
+            currentBlockId.accumulateAndGet(next.key.blockId, EnsureLongIncrementAccumulator.INSTANCE);
         }
     }
 
diff --git a/herddb-core/src/main/java/herddb/index/brin/BlockRangeIndexMetadata.java b/herddb-core/src/main/java/herddb/index/brin/BlockRangeIndexMetadata.java
index 4e0709a1..abf9aea5 100644
--- a/herddb-core/src/main/java/herddb/index/brin/BlockRangeIndexMetadata.java
+++ b/herddb-core/src/main/java/herddb/index/brin/BlockRangeIndexMetadata.java
@@ -40,27 +40,33 @@ public class BlockRangeIndexMetadata<K extends Comparable<K>> {
 
     public static class BlockMetadata<K extends Comparable<K>> {
 
+        public static final byte HEAD = 0b01;
+        public static final byte TAIL = 0b10;
+
         final boolean headBlock;
         final K firstKey;
-        final int blockId;
+        final long blockId;
         final long size;
         final long pageId;
+        Long nextBlockId;
 
-        public BlockMetadata(int blockId, long size, long pageId) {
-            this.headBlock = true;
-            this.firstKey = null;
-            this.blockId = blockId;
-            this.size = size;
-            this.pageId = pageId;
-        }
+//        public BlockMetadata(long blockId, long size, long pageId, Long nextBlockId) {
+//            this.headBlock = true;
+//            this.firstKey = null;
+//            this.blockId = blockId;
+//            this.size = size;
+//            this.pageId = pageId;
+//            this.nextBlockId = nextBlockId;
+//        }
 
-        public BlockMetadata(K firstKey, int blockId, long size, long pageId) {
+        public BlockMetadata(K firstKey, long blockId, long size, long pageId, Long nextBlockId) {
 
             this.headBlock = false;
             this.firstKey = firstKey;
             this.blockId = blockId;
             this.size = size;
             this.pageId = pageId;
+            this.nextBlockId = nextBlockId;
         }
 
         public long getSize() {
@@ -79,9 +85,13 @@ public class BlockRangeIndexMetadata<K extends Comparable<K>> {
             return headBlock;
         }
 
-        public int getBlockId() {
+        public long getBlockId() {
             return blockId;
         }
 
+        public Long getNextBlockId() {
+            return nextBlockId;
+        }
+
     }
 }


Test patch:
diff --git a/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexConcurrentTest.java b/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexConcurrentTest.java
index 7ef3f112..4b9ad95f 100644
--- a/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexConcurrentTest.java
+++ b/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexConcurrentTest.java
@@ -22,17 +22,21 @@ package herddb.index.brin;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
 import herddb.core.RandomPageReplacementPolicy;
 import herddb.index.brin.BlockRangeIndex.Block;
+import herddb.utils.SizeAwareObject;
 import herddb.utils.Sized;
 
 /**
@@ -44,187 +48,241 @@ public class BlockRangeIndexConcurrentTest {
 
     @Test
     public void testConcurrentWrites() throws Exception {
+
         int testSize = 10000;
         int parallelism = 10;
         BlockRangeIndex<Sized<Integer>, Sized<String>> index =
                 new BlockRangeIndex<>(1024, new RandomPageReplacementPolicy(3));
-//        PagedBlockRangeIndex<Sized<Integer>, Sized<String>> index =
-//                new PagedBlockRangeIndex<>(1024, new ClockProPolicy(3));
-//        PagedBlockRangeIndex<Sized<Integer>, Sized<String>> index =
-//                new PagedBlockRangeIndex<>(1024, new ClockAdaptiveReplacement(3));
-        ExecutorService threadpool = Executors.newFixedThreadPool(parallelism);
-        CountDownLatch l = new CountDownLatch(testSize);
 
         try {
-            for (int i = 0; i < testSize; i++) {
-                int _i = i;
-                threadpool.submit(() -> {
-                    try {
-                        index.put(Sized.valueOf(_i), Sized.valueOf("a" + _i));
-                        l.countDown();
-                    } catch (RuntimeException e) {
-                        e.printStackTrace();
-                    }
-                });
-            }
-        } finally {
-            threadpool.shutdown();
-        }
-
-        assertTrue(l.await(10, TimeUnit.SECONDS));
 
-        index.checkpoint();
+            ExecutorService threadpool = Executors.newFixedThreadPool(parallelism);
+            CountDownLatch l = new CountDownLatch(testSize);
+
+            try {
+                for (int i = 0; i < testSize; i++) {
+                    int _i = i;
+                    threadpool.submit(() -> {
+                        try {
+                            index.put(Sized.valueOf(_i), Sized.valueOf("a" + _i));
+                            l.countDown();
+                        } catch (RuntimeException e) {
+                            e.printStackTrace();
+                        }
+                    });
+                }
+            } finally {
+                threadpool.shutdown();
+            }
 
-        dumpIndex(index);
-        verifyIndex(index);
-        List<Sized<String>> result = index.lookUpRange(Sized.valueOf(0), Sized.valueOf(testSize + 1));
-        for (Sized<String> res : result) {
-            System.out.println("res " + res.dummy);
-        }
+            assertTrue(l.await(10, TimeUnit.SECONDS));
 
-        for (int i = 0; i < testSize; i++) {
-            assertTrue("cannot find " + i, index.containsKey(Sized.valueOf(i)));
-        }
-    }
+            index.checkpoint();
 
-    private void dumpIndex(BlockRangeIndex<?, ?> index) {
-        for (Block b : index.getBlocks().values()) {
-            System.out.println("BLOCK " + b);
-        }
-    }
+            dumpIndex(index);
+            verifyIndex(index);
 
-    private void verifyIndex(BlockRangeIndex<Sized<Integer>, Sized<String>> index) {
-        Integer lastmax = null;
-        for (Block b : index.getBlocks().values()) {
-            if (b.key == BlockRangeIndex.BlockStartKey.HEAD_KEY.HEAD_KEY) {
-                System.out.println("check block " + lastmax + " -> -inf");
-            } else {
-                System.out.println("check block " + lastmax + " -> " + ((Sized<Integer>) b.key.minKey).dummy);
+            List<Sized<String>> result = index.lookUpRange(Sized.valueOf(0), Sized.valueOf(testSize + 1));
+            for (Sized<String> res : result) {
+                System.out.println("res " + res.dummy);
             }
 
-            /* Forcefully load the block to check internal data */
-            b.ensureBlockLoaded();
-
-            if (b.values.isEmpty() && index.getBlocks().size() != 1 && b.key != BlockRangeIndex.BlockStartKey.HEAD_KEY) {
-                fail("non head of degenerate tree empty block");
+            for (int i = 0; i < testSize; i++) {
+                assertTrue("cannot find " + i, index.containsKey(Sized.valueOf(i)));
             }
 
-            if (b.values.isEmpty()) {
-                if (index.getBlocks().size() != 1 && b.key != BlockRangeIndex.BlockStartKey.HEAD_KEY) {
-                    fail("non head of degenerate tree empty block");
-                }
-            } else {
-                if (lastmax == null) {
-                    lastmax = ((Sized<Integer>)b.values.lastKey()).dummy;
-                } else {
-                    Integer entryMin = ((Sized<Integer>)b.values.firstKey()).dummy;
-                    Integer entryMax = ((Sized<Integer>)b.values.lastKey()).dummy;
-                    if (entryMin < lastmax) {
-                        fail(entryMin + " < " + lastmax);
-                    }
-                    lastmax = entryMax;
-                }
-            }
+        } catch (Exception | AssertionError e) {
+
+            deepDumpIndex(index);
+            throw e;
         }
     }
 
     @Test
     public void testConcurrentReadsWritesWithSplits() throws Exception {
+
         int testSize = 10000;
         int parallelism = 10;
-        BlockRangeIndex<Sized<Integer>, Sized<String>> index =
-                new BlockRangeIndex<>(1024, new RandomPageReplacementPolicy(3));
+        BlockRangeIndex<Sized<Integer>, Sized<String>> index = new BlockRangeIndex<>(1024,
+                new RandomPageReplacementPolicy(3));
         ExecutorService threadpool = Executors.newFixedThreadPool(parallelism);
-        CountDownLatch l = new CountDownLatch(testSize);
-        ConcurrentLinkedQueue<Sized<String>> results = new ConcurrentLinkedQueue<>();
+
         try {
-            for (int i = 0; i < testSize; i++) {
-                int _i = i;
-                threadpool.submit(() -> {
-                    try {
+
+            ConcurrentLinkedQueue<Sized<String>> results = new ConcurrentLinkedQueue<>();
+            List<Future<?>> jobs = new ArrayList<>(testSize);
+
+            try {
+                for (int i = 0; i < testSize; i++) {
+                    int _i = i;
+                    jobs.add(threadpool.submit(() -> {
                         index.put(Sized.valueOf(_i), Sized.valueOf("a" + _i));
                         List<Sized<String>> search = index.search(Sized.valueOf(_i));
                         results.addAll(search);
                         if (search.isEmpty()) {
+                            search = index.search(Sized.valueOf(_i));
                             throw new IllegalStateException("Empty Search! i " + _i);
                         }
-                        l.countDown();
-                    } catch (RuntimeException e) {
-                        e.printStackTrace();
-                    }
-                });
+                    }));
+                }
+            } finally {
+                threadpool.shutdown();
             }
-        } finally {
-            threadpool.shutdown();
-        }
 
-        assertTrue(l.await(10, TimeUnit.SECONDS));
+            for (Future<?> job : jobs) {
+                /* Job exception rethrow */
+                try {
+                    job.get(10, TimeUnit.SECONDS);
+                } catch (ExecutionException e) {
+                    throw (Exception) e.getCause();
+                }
+            }
 
-        index.checkpoint();
+            index.checkpoint();
 
-        dumpIndex(index);
-        verifyIndex(index);
-        List<Sized<String>> result = index.lookUpRange(Sized.valueOf(0), Sized.valueOf(testSize + 1));
-        for (Sized<String> res : result) {
-            System.out.println("res " + res.dummy);
-        }
+            dumpIndex(index);
+            verifyIndex(index);
+            List<Sized<String>> result = index.lookUpRange(Sized.valueOf(0), Sized.valueOf(testSize + 1));
+            for (Sized<String> res : result) {
+                System.out.println("res " + res.dummy);
+            }
+
+            for (int i = 0; i < testSize; i++) {
+                assertTrue("cannot find " + i, index.containsKey(Sized.valueOf(i)));
+                assertTrue("cannot find a" + i, results.contains(Sized.valueOf("a" + i)));
+            }
 
-        for (int i = 0; i < testSize; i++) {
-            assertTrue("cannot find " + i, index.containsKey(Sized.valueOf(i)));
-            assertTrue("cannot find a" + i, results.contains(Sized.valueOf("a" + i)));
+        } catch (Exception | AssertionError e) {
+
+            deepDumpIndex(index);
+            throw e;
         }
     }
 
     @Test
     public void testConcurrentReadsWritesDeletesWithSplits() throws Exception {
+
         int testSize = 10000;
         int parallelism = 10;
         BlockRangeIndex<Sized<Integer>, Sized<String>> index =
                 new BlockRangeIndex<>(1024, new RandomPageReplacementPolicy(3));
-        ExecutorService threadpool = Executors.newFixedThreadPool(parallelism);
-        CountDownLatch l = new CountDownLatch(testSize);
-        ConcurrentLinkedQueue<Sized<String>> afterInsertSearch = new ConcurrentLinkedQueue<>();
-        ConcurrentLinkedQueue<Sized<String>> afterDeleteSearch = new ConcurrentLinkedQueue<>();
+
         try {
-            for (int i = 0; i < testSize; i++) {
-                int _i = i;
-                threadpool.submit(() -> {
-                    try {
-                        index.put(Sized.valueOf(_i), Sized.valueOf("a" + _i));
-                        List<Sized<String>> search = index.search(Sized.valueOf(_i));
-                        afterInsertSearch.addAll(search);
-                        if (search.isEmpty()) {
-                            throw new IllegalStateException("Empty Search! i " + _i);
+
+            ExecutorService threadpool = Executors.newFixedThreadPool(parallelism);
+            CountDownLatch l = new CountDownLatch(testSize);
+            ConcurrentLinkedQueue<Sized<String>> afterInsertSearch = new ConcurrentLinkedQueue<>();
+            ConcurrentLinkedQueue<Sized<String>> afterDeleteSearch = new ConcurrentLinkedQueue<>();
+            List<Future<?>> jobs = new ArrayList<>(testSize);
+
+            try {
+                for (int i = 0; i < testSize; i++) {
+                    int _i = i;
+                    jobs.add(threadpool.submit(() -> {
+                        try {
+                            index.put(Sized.valueOf(_i), Sized.valueOf("a" + _i));
+                            List<Sized<String>> search = index.search(Sized.valueOf(_i));
+                            afterInsertSearch.addAll(search);
+                            if (search.isEmpty()) {
+                                throw new IllegalStateException("Empty Search! i " + _i);
+                            }
+
+                            index.delete(Sized.valueOf(_i), Sized.valueOf("a" + _i));
+                            List<Sized<String>> search2 = index.search(Sized.valueOf(_i));
+                            afterDeleteSearch.addAll(search2);
+                            l.countDown();
+                        } catch (Throwable t) {
+                            t.printStackTrace();
                         }
+                    }));
+                }
+            } finally {
+                threadpool.shutdown();
+            }
 
-                        index.delete(Sized.valueOf(_i), Sized.valueOf("a" + _i));
-                        List<Sized<String>> search2 = index.search(Sized.valueOf(_i));
-                        afterDeleteSearch.addAll(search2);
-                        l.countDown();
-                    } catch (Throwable t) {
-                        t.printStackTrace();
-                    }
-                });
+            for(Future<?> job : jobs) {
+                /* Job exception rethrow */
+                try {
+                    job.get(10, TimeUnit.SECONDS);
+                } catch (ExecutionException e) {
+                    throw (Exception) e.getCause();
+                }
             }
-        } finally {
-            threadpool.shutdown();
+
+            index.checkpoint();
+            dumpIndex(index);
+
+            verifyIndex(index);
+            List<Sized<String>> lookupAfterDeletion = index.lookUpRange(Sized.valueOf(0), Sized.valueOf(testSize + 1));
+            assertTrue(lookupAfterDeletion.isEmpty());
+
+            for (int i = 0; i < testSize; i++) {
+                assertTrue("cannot find a" + i, afterInsertSearch.contains(Sized.valueOf("a" + i)));
+            }
+            assertTrue(afterDeleteSearch.isEmpty());
+
+        } catch (Exception | AssertionError e) {
+
+            deepDumpIndex(index);
+            throw e;
         }
 
-        assertTrue(l.await(10, TimeUnit.SECONDS));
 
-        index.checkpoint();
+    }
+
 
-        dumpIndex(index);
-        verifyIndex(index);
-        List<Sized<String>> lookupAfterDeletion = index.lookUpRange(Sized.valueOf(0), Sized.valueOf(testSize + 1));
-        assertTrue(lookupAfterDeletion.isEmpty());
+    private <X extends Comparable<X> & SizeAwareObject> void dumpIndex(BlockRangeIndex<X, ?> index) {
+        for (Block<X,?> b : index.getBlocks().values()) {
+            System.out.println("BLOCK " + b);
+        }
+    }
+
+    private <X extends Comparable<X> & SizeAwareObject> void deepDumpIndex(BlockRangeIndex<X, ?> index) {
+        for (Block<X,?> block : index.getBlocks().values()) {
+            System.out.println("------------------------------------------");
+            System.out.println("BLOCK " + block);
+            System.out.println("------------------------------------------");
+
+            /* Forcefully load the block to check internal data */
+            block.ensureBlockLoaded();
 
-        System.out.println(afterInsertSearch);
-        for (int i = 0; i < testSize; i++) {
-            assertTrue("cannot find a" + i, afterInsertSearch.contains(Sized.valueOf("a" + i)));
+            block.values.forEach((k,v) -> System.out.println(k + ": " + v));
         }
-        assertTrue(afterDeleteSearch.isEmpty());
+    }
+
+    private void verifyIndex(BlockRangeIndex<Sized<Integer>, Sized<String>> index) {
+        Integer lastmax = null;
+        for (Block<Sized<Integer>, Sized<String>> b : index.getBlocks().values()) {
+            if (b.key == BlockRangeIndex.BlockStartKey.HEAD_KEY.HEAD_KEY) {
+                System.out.println("check block " + lastmax + " -> -inf");
+            } else {
+                System.out.println("check block " + lastmax + " -> " + b.key.minKey.dummy);
+            }
+
+            /* Forcefully load the block to check internal data */
+            b.ensureBlockLoaded();
+
+            if (b.values.isEmpty() && index.getBlocks().size() != 1 && b.key != BlockRangeIndex.BlockStartKey.HEAD_KEY) {
+                fail("non head of degenerate tree empty block");
+            }
 
+            if (b.values.isEmpty()) {
+                if (index.getBlocks().size() != 1 && b.key != BlockRangeIndex.BlockStartKey.HEAD_KEY) {
+                    fail("non head of degenerate tree empty block");
+                }
+            } else {
+                if (lastmax == null) {
+                    lastmax = b.values.lastKey().dummy;
+                } else {
+                    Integer entryMin = b.values.firstKey().dummy;
+                    Integer entryMax = b.values.lastKey().dummy;
+                    if (entryMin < lastmax) {
+                        fail(entryMin + " < " + lastmax);
+                    }
+                    lastmax = entryMax;
+                }
+            }
+        }
     }
 
 }
diff --git a/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexStorageTest.java b/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexStorageTest.java
index ef39bd36..fd7f6857 100644
--- a/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexStorageTest.java
+++ b/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexStorageTest.java
@@ -21,16 +21,19 @@ package herddb.index.brin;
 
 import static org.junit.Assert.assertEquals;
 
-import java.io.IOException;
+import java.util.AbstractMap.SimpleEntry;
+import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentNavigableMap;
 
+import org.junit.Assert;
 import org.junit.Test;
 
 import herddb.core.PageReplacementPolicy;
 import herddb.core.RandomPageReplacementPolicy;
+import herddb.index.brin.BlockRangeIndex.Block;
+import herddb.index.brin.BlockRangeIndex.BlockStartKey;
 import herddb.utils.Sized;
 
 /**
@@ -40,32 +43,12 @@ import herddb.utils.Sized;
 public class BlockRangeIndexStorageTest {
 
     @Test
-    public void testSimple() throws Exception {
+    public void testSimpleReload() throws Exception {
 
-        IndexDataStorage<Sized<Integer>, Sized<String>> storage =
-                new IndexDataStorage<Sized<Integer>, Sized<String>>() {
-
-            AtomicLong newPageId = new AtomicLong();
-
-            private ConcurrentHashMap<Long, List<Map.Entry<Sized<Integer>, Sized<String>>>> pages =
-                    new ConcurrentHashMap<>();
-
-            @Override
-            public List<Map.Entry<Sized<Integer>, Sized<String>>> loadDataPage(long pageId) throws IOException {
-                return pages.get(pageId);
-            }
-
-            @Override
-            public long createDataPage(List<Map.Entry<Sized<Integer>, Sized<String>>> values) throws IOException {
-                long newid = newPageId.incrementAndGet();
-                pages.put(newid, values);
-                return newid;
-            }
-
-        };
+        PageReplacementPolicy policy = new RandomPageReplacementPolicy(10);
+        IndexDataStorage<Sized<Integer>, Sized<String>> storage = new MemoryIndexDataStorage<>();
 
-        BlockRangeIndex<Sized<Integer>, Sized<String>> index =
-                new BlockRangeIndex<>(400, new RandomPageReplacementPolicy(10), storage);
+        BlockRangeIndex<Sized<Integer>, Sized<String>> index = new BlockRangeIndex<>(400, policy, storage);
 
         index.put(Sized.valueOf(1), Sized.valueOf("a"));
         index.put(Sized.valueOf(2), Sized.valueOf("b"));
@@ -86,31 +69,74 @@ public class BlockRangeIndexStorageTest {
     }
 
     @Test
-    public void testUnload() throws Exception {
+    public void testReloadAfterBlockDeletion() throws Exception {
+
+        PageReplacementPolicy policy = new RandomPageReplacementPolicy(10);
+        IndexDataStorage<Sized<Integer>, Sized<Integer>> storage = new MemoryIndexDataStorage<>();
+
+        BlockRangeIndex<Sized<Integer>, Sized<Integer>> index = new BlockRangeIndex<>(400, policy, storage);
+
+        int i = 0;
+        do {
+            Sized<Integer> si = Sized.valueOf(i++);
+            index.put(si, si);
+        } while(index.getNumBlocks() < 4);
+
+        /* First checkpoint after insertion */
+        index.checkpoint();
+
+
+        /* Now we empty middle blocks */
+
+        /* Map without first key */
+        ConcurrentNavigableMap<BlockStartKey<Sized<Integer>>, Block<Sized<Integer>, Sized<Integer>>> sub =
+                index.getBlocks().tailMap(index.getBlocks().firstKey(), false);
 
-        IndexDataStorage<Sized<Integer>, Sized<String>> storage = new IndexDataStorage<Sized<Integer>, Sized<String>>() {
+        /* Second block */
+        Block<Sized<Integer>, Sized<Integer>> second = sub.firstEntry().getValue();
 
-            AtomicLong newPageId = new AtomicLong();
+        /* Map without second key */
+        sub = sub.tailMap(sub.firstKey(), false);
 
-            private ConcurrentHashMap<Long, List<Map.Entry<Sized<Integer>, Sized<String>>>> pages = new ConcurrentHashMap<>();
+        /* Third block */
+        Block<Sized<Integer>, Sized<Integer>> third = sub.firstEntry().getValue();
 
-            @Override
-            public List<Map.Entry<Sized<Integer>, Sized<String>>> loadDataPage(long pageId) throws IOException {
-                return pages.get(pageId);
-            }
+        /* Copy to avoid concurrent modification */
+        List<Entry<Sized<Integer>, Sized<Integer>>> toDelete = new ArrayList<>();
 
-            @Override
-            public long createDataPage(List<Map.Entry<Sized<Integer>, Sized<String>>> values) throws IOException {
-                long newid = newPageId.incrementAndGet();
-                pages.put(newid, values);
-                return newid;
-            }
+        second.values.forEach((k,vl) -> vl.forEach(v -> toDelete.add(new SimpleEntry<>(k,v))));
+        third.values.forEach((k,vl) -> vl.forEach(v -> toDelete.add(new SimpleEntry<>(k,v))));
 
-        };
+        /* Delete blocks 2 and 3 */
+        toDelete.forEach(e -> index.delete(e.getKey(), e.getValue()));
+
+        /* Checkpoint, should remove a block */
+        BlockRangeIndexMetadata<Sized<Integer>> metadata = index.checkpoint();
+
+        assertEquals(2, index.getNumBlocks());
+        assertEquals(index.getNumBlocks(), metadata.getBlocksMetadata().size());
+
+        /* Load a new index from data */
+        BlockRangeIndex<Sized<Integer>, Sized<Integer>> indexAfterBoot =
+                new BlockRangeIndex<>(1024, new RandomPageReplacementPolicy(10), storage);
+
+        indexAfterBoot.boot(metadata);
+
+        /* Check data equality between new and old index */
+        index.getBlocks().forEach((f,b) -> b.values.forEach((k,vl) -> {
+            List<Sized<Integer>> search = indexAfterBoot.search(k);
+            Assert.assertEquals(vl, search);
+        }));
+
+    }
+
+    @Test
+    public void testUnload() throws Exception {
 
         PageReplacementPolicy policy = new RandomPageReplacementPolicy(10);
-        BlockRangeIndex<Sized<Integer>, Sized<String>> index =
-                new BlockRangeIndex<>(1024, policy, storage);
+        IndexDataStorage<Sized<Integer>, Sized<String>> storage = new MemoryIndexDataStorage<>();
+
+        BlockRangeIndex<Sized<Integer>, Sized<String>> index = new BlockRangeIndex<>(1024, policy, storage);
         for (int i = 0; i < 100; i++) {
             index.put(Sized.valueOf(i), Sized.valueOf("a"));
         }
diff --git a/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexTest.java b/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexTest.java
index 82b665d8..a3ee4f0a 100644
--- a/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexTest.java
+++ b/herddb-core/src/test/java/herddb/index/brin/BlockRangeIndexTest.java
@@ -55,6 +55,30 @@ public class BlockRangeIndexTest {
         assertEquals(2, index.getNumBlocks());
     }
 
+    @Test
+    public void testNotNullNextAfterSplit() {
+
+        BlockRangeIndex<Sized<Integer>, Sized<Integer>> index =
+                new BlockRangeIndex<>(400, new RandomPageReplacementPolicy(10));
+
+        int i = 0;
+        do {
+            /*
+             * Add elements at index head to generate 3 blocks and beeing sure that just the
+             * last block has a null next and not because we split only the last block with
+             * null next. Inserting top elements we split head block every time and such
+             * block should have null next just when is the only one block.
+             */
+            Sized<Integer> si = Sized.valueOf(--i);
+            index.put(si, si);
+        } while (index.getNumBlocks() < 3);
+
+
+        int nulls = index.getBlocks().values().stream().mapToInt(b -> b.next == null ? 1 : 0).sum();
+
+        Assert.assertEquals(1, nulls);
+    }
+
     @Test
     public void testRemoveHead() {
         BlockRangeIndex<Sized<Integer>, Sized<String>> index =