add list index that will keep x entries per page, basis for plist that is more space efficient

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1127543 13f79535-47bb-0310-9956-ffa450edef68

kahadb/src/main/java/org/apache/kahadb/index/ListIndex.java

@@ -0,0 +1,257 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kahadb.index;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.kahadb.page.Page;
+import org.apache.kahadb.page.PageFile;
+import org.apache.kahadb.page.Transaction;
+import org.apache.kahadb.util.Marshaller;
+
+public class ListIndex<Key,Value> implements Index<Key,Value> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ListIndex.class);
+
+    protected PageFile pageFile;
+    protected long headPageId;
+    protected long tailPageId;
+    private long size;
+
+    protected AtomicBoolean loaded = new AtomicBoolean();
+
+    private final ListNode.Marshaller<Key, Value> marshaller = new ListNode.Marshaller<Key, Value>(this);
+    private Marshaller<Key> keyMarshaller;
+    private Marshaller<Value> valueMarshaller;
+
+    public ListIndex(PageFile pageFile, long rootPageId) {
+        this.pageFile = pageFile;
+        this.headPageId = rootPageId;
+    }
+
+    synchronized public void load(Transaction tx) throws IOException {
+        if (loaded.compareAndSet(false, true)) {
+            LOG.debug("loading");
+            if( keyMarshaller == null ) {
+                throw new IllegalArgumentException("The key marshaller must be set before loading the ListIndex");
+            }
+            if( valueMarshaller == null ) {
+                throw new IllegalArgumentException("The value marshaller must be set before loading the ListIndex");
+            }
+            
+            final Page<ListNode<Key,Value>> p = tx.load(headPageId, null);
+            if( p.getType() == Page.PAGE_FREE_TYPE ) {
+                 // Need to initialize it..
+                ListNode<Key, Value> root = createNode(p, null);
+                storeNode(tx, root, true);
+                tailPageId = headPageId;
+            } else {
+                ListNode<Key, Value> node = loadNode(tx, headPageId, null);
+                size += node.size(tx);
+                while (node.getNext() != -1) {
+                    node = loadNode(tx, node.getNext(), node);
+                    size += node.size(tx);
+                    tailPageId = node.getPageId();
+                }
+            }
+        }
+    }
+    
+    synchronized public void unload(Transaction tx) {
+        if (loaded.compareAndSet(true, false)) {
+        }    
+    }
+    
+    protected ListNode<Key,Value> getHead(Transaction tx) throws IOException {
+        return loadNode(tx, headPageId, null);
+    }
+
+    protected ListNode<Key,Value> getTail(Transaction tx) throws IOException {
+        return loadNode(tx, tailPageId, null);
+    }
+
+    synchronized public boolean containsKey(Transaction tx, Key key) throws IOException {
+        assertLoaded();
+        for (Iterator<Map.Entry<Key,Value>> iterator = iterator(tx); iterator.hasNext(); ) {
+            Map.Entry<Key,Value> candidate = iterator.next();
+            if (key.equals(candidate.getKey())) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    synchronized public Value get(Transaction tx, Key key) throws IOException {
+        assertLoaded();
+        for (Iterator<Map.Entry<Key,Value>> iterator = iterator(tx); iterator.hasNext(); ) {
+            Map.Entry<Key,Value> candidate = iterator.next();
+            if (key.equals(candidate.getKey())) {
+                return candidate.getValue();
+            }
+        }
+        return null;
+    }
+
+    /**
+      * appends to the list
+     * @return null
+     */
+    synchronized public Value put(Transaction tx, Key key, Value value) throws IOException {
+        return add(tx, key, value);
+    }
+
+    synchronized public Value add(Transaction tx, Key key, Value value) throws IOException {
+        assertLoaded();
+        getTail(tx).put(tx, key, value);
+        size ++;
+        return null;
+    }
+
+    synchronized public Value addFirst(Transaction tx, Key key, Value value) throws IOException {
+        assertLoaded();
+        getHead(tx).addFirst(tx, key, value);
+        size++;
+        return null;
+    }
+
+    synchronized public Value remove(Transaction tx, Key key) throws IOException {
+        assertLoaded();
+        for (Iterator<Map.Entry<Key,Value>> iterator = iterator(tx); iterator.hasNext(); ) {
+            Map.Entry<Key,Value> candidate = iterator.next();
+            if (key.equals(candidate.getKey())) {
+                iterator.remove();
+                return candidate.getValue();
+            }
+        }
+        return null;
+    }
+
+    public void onRemove() {
+        size--;
+    }
+
+    public boolean isTransient() {
+        return false;
+    }
+
+    synchronized public void clear(Transaction tx) throws IOException {
+        for (Iterator<ListNode<Key,Value>> iterator = listNodeIterator(tx); iterator.hasNext(); ) {
+            ListNode<Key,Value>candidate = iterator.next();
+            candidate.clear(tx);
+        }
+        size = 0;
+    }
+
+    synchronized public Iterator<ListNode<Key, Value>> listNodeIterator(Transaction tx) throws IOException {
+        return getHead(tx).listNodeIterator(tx);
+    }
+
+    synchronized public boolean isEmpty(final Transaction tx) throws IOException {
+        return getHead(tx).isEmpty(tx);
+    }
+
+    synchronized public Iterator<Map.Entry<Key,Value>> iterator(final Transaction tx) throws IOException {
+        return getHead(tx).iterator(tx);
+    }
+    
+    synchronized public Iterator<Map.Entry<Key,Value>> iterator(final Transaction tx, int initialPosition) throws IOException {
+        return getHead(tx).iterator(tx, initialPosition);
+    }
+
+    synchronized public Map.Entry<Key,Value> getFirst(Transaction tx) throws IOException {
+        return getHead(tx).getFirst(tx);
+    }
+
+    synchronized public Map.Entry<Key,Value> getLast(Transaction tx) throws IOException {
+        return getTail(tx).getLast(tx);
+    }
+
+    private void assertLoaded() throws IllegalStateException {
+        if( !loaded.get() ) {
+            throw new IllegalStateException("TheListIndex is not loaded");
+        }
+    }
+
+    ListNode<Key,Value> loadNode(Transaction tx, long pageId, ListNode<Key,Value> parent) throws IOException {
+        Page<ListNode<Key,Value>> page = tx.load(pageId, marshaller);
+        ListNode<Key, Value> node = page.get();
+        node.setPage(page);
+        node.setParent(parent);
+        return node;
+    }
+
+    ListNode<Key,Value> createNode(Page<ListNode<Key,Value>> p, ListNode<Key,Value> parent) throws IOException {
+        ListNode<Key,Value> node = new ListNode<Key,Value>(this);
+        node.setPage(p);
+        node.setParent(parent);
+        node.setEmpty();
+        p.set(node);
+        return node;
+    }
+
+    ListNode<Key,Value> createNode(Transaction tx, ListNode<Key,Value> parent) throws IOException {
+        Page<ListNode<Key,Value>> page = tx.load(tx.<Object>allocate(1).getPageId(), marshaller);
+        ListNode<Key,Value> node = new ListNode<Key,Value>(this);
+        node.setPage(page);
+        node.setParent(parent);
+        node.setEmpty();
+        page.set(node);
+        return node;
+    }
+
+    void storeNode(Transaction tx, ListNode<Key,Value> node, boolean overflow) throws IOException {
+        tx.store(node.getPage(), marshaller, overflow);
+    }
+        
+    public PageFile getPageFile() {
+        return pageFile;
+    }
+    public long getHeadPageId() {
+        return headPageId;
+    }
+
+    public void setHeadPageId(long headPageId) {
+        this.headPageId = headPageId;
+    }
+
+    public Marshaller<Key> getKeyMarshaller() {
+        return keyMarshaller;
+    }
+    public void setKeyMarshaller(Marshaller<Key> keyMarshaller) {
+        this.keyMarshaller = keyMarshaller;
+    }
+
+    public Marshaller<Value> getValueMarshaller() {
+        return valueMarshaller;
+    }
+    public void setValueMarshaller(Marshaller<Value> valueMarshaller) {
+        this.valueMarshaller = valueMarshaller;
+    }
+
+    public void setTailPageId(long tailPageId) {
+        this.tailPageId = tailPageId;
+    }
+
+    public long size() {
+        return size;
+    }
+}

kahadb/src/main/java/org/apache/kahadb/index/ListNode.java

@@ -0,0 +1,430 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kahadb.index;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.NoSuchElementException;
+import javax.swing.plaf.basic.BasicInternalFrameTitlePane;
+import org.apache.kahadb.index.BTreeIndex.Prefixer;
+import org.apache.kahadb.page.Page;
+import org.apache.kahadb.page.Transaction;
+import org.apache.kahadb.util.LinkedNode;
+import org.apache.kahadb.util.LinkedNodeList;
+import org.apache.kahadb.util.VariableMarshaller;
+import sun.plugin.dom.exception.InvalidStateException;
+import sun.tools.tree.ReturnStatement;
+import sun.util.resources.CurrencyNames_th_TH;
+
+/**
+ * The ListNode class represents a node in the List object graph.  It is stored in
+ * one overflowing Page of a PageFile.
+ */
+public final class ListNode<Key,Value> {
+
+    // The index that this node is part of.
+    private final ListIndex<Key,Value> index;
+    // The parent node or null if this is the root node of the List
+    private ListNode<Key,Value> parent;
+    // The page associated with this node
+    private Page<ListNode<Key,Value>> page;
+
+    protected LinkedNodeList<KeyValueEntry<Key, Value>> entries = new LinkedNodeList<KeyValueEntry<Key, Value>>();
+
+    // The next page after this one.
+    private long next = -1;
+
+    public int size(Transaction tx) {
+        return entries.size();
+    }
+
+    static final class KeyValueEntry<Key, Value> extends LinkedNode<KeyValueEntry<Key, Value>> implements Entry<Key, Value>
+    {
+        private final Key key;
+        private final Value value;
+
+        public KeyValueEntry(Key key, Value value) {
+            this.key = key;
+            this.value = value;
+        }
+
+        public Key getKey() {
+            return key;
+        }
+
+        public Value getValue() {
+            return value;
+        }
+
+        public Value setValue(Value value) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public String toString() {
+            return "{" + key + ":" + value + "}";
+        }
+    }
+
+    private final class ListNodeIterator implements Iterator<ListNode<Key,Value>> {
+
+        private final Transaction tx;
+        ListNode<Key,Value> nextEntry;
+
+        private ListNodeIterator(Transaction tx, ListNode<Key,Value> current) throws IOException {
+            this.tx = tx;
+            nextEntry = current;
+        }
+
+        public boolean hasNext() {
+            return nextEntry !=null;
+        }
+
+        public ListNode<Key,Value> next() {
+            ListNode<Key,Value> current = nextEntry;
+            if( nextEntry !=null ) {
+                if (nextEntry.next != -1) {
+                    try {
+                        nextEntry = index.loadNode(tx, current.next, current);
+                    } catch (IOException unexpected) {
+                        InvalidStateException e = new InvalidStateException("failed to load next: " + current.next + ", reason: " + unexpected.getLocalizedMessage());
+                        e.initCause(unexpected);
+                        throw e;
+                    }
+                } else {
+                    nextEntry = null;
+                }
+            }
+            return current;
+        }
+
+        public void remove() {
+            throw new UnsupportedOperationException();
+        }
+    }
+
+    private final class ListIterator implements Iterator<Entry<Key, Value>> {
+
+        private final Transaction tx;
+        ListNode<Key,Value> current;
+        KeyValueEntry<Key, Value> nextEntry;
+        KeyValueEntry<Key, Value>  toRemove;
+
+        private ListIterator(Transaction tx, ListNode<Key,Value> current, int nextIndex) throws IOException {
+            this.tx = tx;
+            this.current = current;
+            nextEntry = current.entries.getHead();
+            if (nextIndex > 0) {
+                for (int i=0; i<nextIndex; i++) {
+                    nextEntry = nextEntry.getNext();
+                    if (nextEntry == null) {
+                        if (!nextFromNextListNode())
+                            throw new NoSuchElementException("Index out of range: " + nextIndex);
+                        }
+                    }
+                }
+            }
+
+        private boolean nextFromNextListNode() {
+            boolean haveNext = false;
+            if (current.getNext() != -1) {
+                try {
+                    current = index.loadNode(tx, current.getNext(), current);
+                } catch (IOException unexpected) {
+                    NoSuchElementException e = new NoSuchElementException(unexpected.getLocalizedMessage());
+                    e.initCause(unexpected);
+                    throw e;
+                }
+                nextEntry = current.entries.getHead();
+                haveNext = nextEntry != null;
+            }
+            return haveNext;
+        }
+
+        public boolean hasNext() {
+            return nextEntry !=null || nextFromNextListNode();
+        }
+
+        public Entry<Key, Value> next() {
+            if( nextEntry !=null ) {
+                toRemove = nextEntry;
+                nextEntry=toRemove.getNext();
+                return toRemove;
+            } else {
+                throw new NoSuchElementException();
+            }
+        }
+
+        public void remove() {
+            if (toRemove == null) {
+                throw new InvalidStateException("can only remove once, call next again");
+            }
+            try {
+                doRemove(tx, current, toRemove);
+                index.onRemove();
+                toRemove = null;
+            } catch (IOException unexpected) {
+                InvalidStateException e = new InvalidStateException(unexpected.getLocalizedMessage());
+                e.initCause(unexpected);
+                throw e;
+            }
+        }
+    }
+
+    /**
+     * The Marshaller is used to store and load the data in the ListNode into a Page.
+     *
+     * @param <Key>
+     * @param <Value>
+     */
+    static public class Marshaller<Key,Value> extends VariableMarshaller<ListNode<Key,Value>> {
+        private final ListIndex<Key,Value> index;
+
+        public Marshaller(ListIndex<Key,Value> index) {
+            this.index = index;
+        }
+
+        public void writePayload(ListNode<Key,Value> node, DataOutput os) throws IOException {
+            // Write the keys
+            short count = (short)node.entries.size(); // cast may truncate value...
+            if( count != node.entries.size() ) {
+                throw new IOException("short over flow, too many entries in list: " + node.entries.size());
+            }
+
+            os.writeShort(count);
+            KeyValueEntry<Key, Value> entry = node.entries.getHead();
+            while (entry != null) {
+                index.getKeyMarshaller().writePayload((Key) entry.getKey(), os);
+                index.getValueMarshaller().writePayload((Value) entry.getValue(), os);
+                entry = entry.getNext();
+            }
+        }
+
+        @SuppressWarnings("unchecked")
+        public ListNode<Key,Value> readPayload(DataInput is) throws IOException {
+            ListNode<Key,Value> node = new ListNode<Key,Value>(index);
+            final short size = is.readShort();
+            for (short i = 0; i < size; i++) {
+                node.entries.addLast(
+                        new KeyValueEntry(index.getKeyMarshaller().readPayload(is),
+                                                     index.getValueMarshaller().readPayload(is)));
+            }
+            return node;
+        }
+    }
+
+    public ListNode(ListIndex<Key, Value> index) {
+        this.index = index;
+    }
+
+    public void setEmpty() {
+    }
+
+    public Value remove(Transaction tx, Key key) throws IOException {
+        Value result = null;
+        KeyValueEntry<Key, Value> entry = entries.getHead();
+        while (entry != null) {
+            if (entry.getKey().equals(key)) {
+                 result = entry.getValue();
+                 doRemove(tx, this, entry);
+                 break;
+            }
+            entry = entry.getNext();
+        }
+        return result;
+    }
+
+    private void doRemove(Transaction tx, ListNode current, KeyValueEntry<Key, Value> entry) throws IOException {
+        entry.unlink();
+        if (current.entries.isEmpty()) {
+                if (current.getPageId() == index.getHeadPageId()) {
+                    if (current.getNext() != -1) {
+                        // new head
+                        index.setHeadPageId(current.getNext());
+                        tx.free(current.getPageId());
+                    }
+                } else {
+                    // need to unlink the node
+                    current.parent.setNext(current.getNext());
+                    tx.free(current.getPageId());
+                    index.storeNode(tx, current.parent, false);
+                }
+        } else {
+            store(tx, true);
+        }
+    }
+
+    public Value put(Transaction tx, Key key, Value value) throws IOException {
+        if (key == null) {
+            throw new IllegalArgumentException("Key cannot be null");
+        }
+        entries.addLast(new KeyValueEntry(key, value));
+        store(tx, false);
+        return null;
+    }
+
+    public Value addFirst(Transaction tx, Key key, Value value) throws IOException {
+        if (key == null) {
+            throw new IllegalArgumentException("Key cannot be null");
+        }
+        entries.addFirst(new KeyValueEntry(key, value));
+        store(tx, true);
+        return null;
+    }
+
+    private void store(Transaction tx, boolean addFirst) throws IOException {
+        try {
+            index.storeNode(tx, this, allowOverflow());
+        } catch ( Transaction.PageOverflowIOException e ) {
+                // If we get an overflow
+                split(tx, addFirst);
+        }
+    }
+
+    private boolean allowOverflow() {
+        return false;
+    }
+
+    private void split(Transaction tx, boolean isAddFirst) throws IOException {
+        ListNode<Key, Value> extension = index.createNode(tx, this);
+        if (isAddFirst) {
+            extension.setEntries(entries.getHead().splitAfter());
+            extension.setNext(this.getNext());
+            this.setNext(extension.getPageId());
+        }  else {
+            index.setTailPageId(extension.getPageId());
+            this.setNext(extension.getPageId());
+            extension.setEntries(entries.getTail().getPrevious().splitAfter());
+        }
+        index.storeNode(tx, this, false);
+        extension.store(tx, isAddFirst);
+    }
+
+    // called after a split
+    private void setEntries(LinkedNodeList<KeyValueEntry<Key, Value>> list) {
+        this.entries = list;
+    }
+
+    public Value get(Transaction tx, Key key) throws IOException {
+        if (key == null) {
+            throw new IllegalArgumentException("Key cannot be null");
+        }
+        Value result = null;
+        KeyValueEntry<Key, Value> nextEntry = entries.getTail();
+        while (nextEntry != null) {
+            if (nextEntry.getKey().equals(key)) {
+                result =  nextEntry.getValue();
+                break;
+            }
+            nextEntry = nextEntry.getPrevious();
+        }
+        return result;
+    }
+
+    public boolean isEmpty(final Transaction tx) throws IOException {
+        return entries.isEmpty();
+    }
+
+    public Entry<Key,Value> getFirst(Transaction tx) throws IOException {
+        return entries.getHead();
+    }
+
+    public Entry<Key,Value> getLast(Transaction tx) throws IOException {
+        return entries.getTail();
+    }
+
+    public Iterator<Entry<Key,Value>> iterator(final Transaction tx, int pos) throws IOException {
+        return new ListIterator(tx, this, pos);
+    }
+
+    public Iterator<Entry<Key,Value>> iterator(final Transaction tx) throws IOException {
+        return new ListIterator(tx, this, 0);
+    }
+
+    Iterator<ListNode<Key, Value>> listNodeIterator(final Transaction tx) throws IOException {
+        return new ListNodeIterator(tx, this);
+    }
+
+    public void clear(Transaction tx) throws IOException {
+        entries.clear();
+        tx.free(this.getPageId());
+    }
+
+    public boolean contains(Transaction tx, Key key) throws IOException {
+        if (key == null) {
+            throw new IllegalArgumentException("Key cannot be null");
+        }
+        boolean found = false;
+        KeyValueEntry<Key, Value> nextEntry = entries.getTail();
+        while (nextEntry != null) {
+            if (nextEntry.getKey().equals(key)) {
+                found = true;
+                break;
+            }
+            nextEntry = nextEntry.getPrevious();
+        }
+        return found;
+    }
+
+    ///////////////////////////////////////////////////////////////////
+    // Implementation methods
+    ///////////////////////////////////////////////////////////////////
+ 
+    public long getPageId() {
+        return page.getPageId();
+    }
+
+    public ListNode<Key, Value> getParent() {
+        return parent;
+    }
+
+    public void setParent(ListNode<Key, Value> parent) {
+        this.parent = parent;
+    }
+
+    public Page<ListNode<Key, Value>> getPage() {
+        return page;
+    }
+
+    public void setPage(Page<ListNode<Key, Value>> page) {
+        this.page = page;
+    }
+
+    public long getNext() {
+        return next;
+    }
+
+    public void setNext(long next) {
+        this.next = next;
+    }
+    
+    @Override
+    public String toString() {
+        return "[ListNode "+ entries.toString() + "]";
+    }
+}
+
+

kahadb/src/main/java/org/apache/kahadb/page/PageFile.java

@@ -634,7 +634,7 @@ public class PageFile {
     }
 
     /**
-     * @param allows you to enable read page caching
+     * @param enablePageCaching allows you to enable read page caching
      */
     public void setEnablePageCaching(boolean enablePageCaching) {
         assertNotLoaded();
@@ -649,7 +649,7 @@ public class PageFile {
     }
 
     /**
-     * @param Sets the maximum number of pages that will get stored in the read page cache.
+     * @param pageCacheSize Sets the maximum number of pages that will get stored in the read page cache.
      */
     public void setPageCacheSize(int pageCacheSize) {
         assertNotLoaded();
@@ -680,6 +680,11 @@ public class PageFile {
         return recoveryFileMinPageCount;
     }
 
+    public long getFreePageCount() {
+        assertLoaded();
+        return freeList.size();
+    }
+
     public void setRecoveryFileMinPageCount(int recoveryFileMinPageCount) {
         assertNotLoaded();
         this.recoveryFileMinPageCount = recoveryFileMinPageCount;
@@ -915,8 +920,6 @@ public class PageFile {
 
     /**
      * 
-     * @param timeout
-     * @param unit
      * @return true if there are still pending writes to do.
      * @throws InterruptedException 
      * @throws IOException 

kahadb/src/main/java/org/apache/kahadb/util/LinkedNode.java

@@ -262,8 +262,6 @@ public class LinkedNode<T extends LinkedNode<T>> {
         // Update all the nodes in the new list so that they know of their new
         // list owner.
         T n = newList.head;
-        newList.size++;
-        list.size--;
         do {
             n.list = newList;
             n = n.next;
@@ -301,8 +299,6 @@ public class LinkedNode<T extends LinkedNode<T>> {
         // Update all the nodes in the new list so that they know of their new
         // list owner.
         T n = newList.head;
-        newList.size++;
-        list.size--;
         do {
             n.list = newList;
             n = n.next;

kahadb/src/main/java/org/apache/kahadb/util/LinkedNodeList.java

@@ -48,7 +48,7 @@ public class LinkedNodeList<T extends LinkedNode<T>> {
     }
 
     public T getTail() {
-        return head.prev;
+        return head != null ? head.prev : null;
     }
     
     public void clear() {

kahadb/src/test/java/org/apache/kahadb/index/ListIndexTest.java

@@ -0,0 +1,326 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kahadb.index;
+
+import java.text.NumberFormat;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Random;
+import org.apache.kahadb.util.LongMarshaller;
+import org.apache.kahadb.util.StringMarshaller;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ListIndexTest extends IndexTestSupport {
+    private static final Logger LOG = LoggerFactory.getLogger(ListIndexTest.class);
+    private NumberFormat nf;
+
+    @Override
+    protected void setUp() throws Exception {
+        super.setUp();
+        nf = NumberFormat.getIntegerInstance();
+        nf.setMinimumIntegerDigits(6);
+        nf.setGroupingUsed(false);
+    }
+
+    @Override
+    protected Index<String, Long> createIndex() throws Exception {
+
+        long id = tx.allocate().getPageId();
+        tx.commit();
+
+        ListIndex<String, Long> index = new ListIndex<String, Long>(pf, id);
+        index.setKeyMarshaller(StringMarshaller.INSTANCE);
+        index.setValueMarshaller(LongMarshaller.INSTANCE);
+
+        return index;
+    }
+
+    public void testSize() throws Exception {
+        createPageFileAndIndex(100);
+
+        ListIndex<String, Long> listIndex = ((ListIndex<String, Long>) this.index);
+        this.index.load(tx);
+        tx.commit();
+
+        int count = 30;
+        tx = pf.tx();
+        doInsert(count);
+        tx.commit();
+        assertEquals("correct size", count, listIndex.size());
+
+        tx = pf.tx();
+        Iterator<Map.Entry<String, Long>> iterator = index.iterator(tx);
+        while (iterator.hasNext()) {
+            iterator.next();
+            iterator.remove();
+            assertEquals("correct size", --count, listIndex.size());
+        }
+        tx.commit();
+
+        count = 30;
+        tx = pf.tx();
+        doInsert(count);
+        tx.commit();
+        assertEquals("correct size", count, listIndex.size());
+
+        tx = pf.tx();
+        listIndex.clear(tx);
+        assertEquals("correct size", 0, listIndex.size());
+        tx.commit();
+    }
+
+    public void testAddFirst() throws Exception {
+        createPageFileAndIndex(100);
+
+        ListIndex<String, Long> listIndex = ((ListIndex<String, Long>) this.index);
+        this.index.load(tx);
+        tx.commit();
+
+        tx = pf.tx();
+
+        // put is add last
+        doInsert(10);
+        listIndex.addFirst(tx, key(10), (long) 10);
+        listIndex.addFirst(tx, key(11), (long) 11);
+
+        tx.commit();
+
+        tx = pf.tx();
+        int counter = 11;
+        Iterator<Map.Entry<String, Long>> iterator = index.iterator(tx);
+        assertEquals(key(counter), iterator.next().getKey());
+        counter--;
+        assertEquals(key(counter), iterator.next().getKey());
+        counter--;
+        int count = 0;
+        while (iterator.hasNext() && count < counter) {
+            Map.Entry<String, Long> entry = (Map.Entry<String, Long>) iterator.next();
+            assertEquals(key(count), entry.getKey());
+            assertEquals(count, (long) entry.getValue());
+            count++;
+        }
+        tx.commit();
+    }
+
+    public void testPruning() throws Exception {
+        createPageFileAndIndex(100);
+
+        ListIndex<String, Long> index = ((ListIndex<String, Long>) this.index);
+
+        this.index.load(tx);
+        tx.commit();
+
+        long pageCount = index.getPageFile().getPageCount();
+        assertEquals(1, pageCount);
+
+        long freePageCount = index.getPageFile().getFreePageCount();
+        assertEquals("No free pages", 0, freePageCount);
+
+        tx = pf.tx();
+        doInsert(20);
+        tx.commit();
+
+        pageCount = index.getPageFile().getPageCount();
+        LOG.info("page count: " + pageCount);
+        assertTrue("used some pages", pageCount > 1);
+
+        tx = pf.tx();
+        // Remove the data.
+        doRemove(20);
+
+        tx.commit();
+
+        freePageCount = index.getPageFile().getFreePageCount();
+
+        LOG.info("FreePage count: " + freePageCount);
+        assertTrue("Some free pages " + freePageCount, freePageCount > 0);
+
+
+        LOG.info("add some more to use up free list");
+        tx = pf.tx();
+        doInsert(20);
+        tx.commit();
+
+        freePageCount = index.getPageFile().getFreePageCount();
+
+        LOG.info("FreePage count: " + freePageCount);
+        assertEquals("no free pages " + freePageCount, 0, freePageCount);
+        assertEquals("Page count is static", pageCount,  index.getPageFile().getPageCount());
+
+        this.index.unload(tx);
+        tx.commit();
+    }
+
+    public void testIterationAddFirst() throws Exception {
+        createPageFileAndIndex(100);
+        ListIndex<String, Long> index = ((ListIndex<String, Long>) this.index);
+        this.index.load(tx);
+        tx.commit();
+
+        tx = pf.tx();
+        final int entryCount = 200;
+        // Insert in reverse order..
+        doInsertReverse(entryCount);
+        this.index.unload(tx);
+        tx.commit();
+        this.index.load(tx);
+        tx.commit();
+
+        int counter = 0;
+        for (Iterator<Map.Entry<String, Long>> i = index.iterator(tx); i.hasNext(); ) {
+            Map.Entry<String, Long> entry = (Map.Entry<String, Long>) i.next();
+            assertEquals(key(counter), entry.getKey());
+            assertEquals(counter, (long) entry.getValue());
+            counter++;
+        }
+         assertEquals("We iterated over all entries", entryCount, counter);
+
+        tx = pf.tx();
+        // Remove the data.
+        doRemove(entryCount);
+        tx.commit();
+
+        this.index.unload(tx);
+        tx.commit();
+    }
+
+
+    public void testIteration() throws Exception {
+        createPageFileAndIndex(100);
+        ListIndex<String, Long> index = ((ListIndex<String, Long>) this.index);
+        this.index.load(tx);
+        tx.commit();
+
+        // Insert in reverse order..
+        final int entryCount = 200;
+        doInsert(entryCount);
+
+        this.index.unload(tx);
+        tx.commit();
+        this.index.load(tx);
+        tx.commit();
+
+        int counter = 0;
+        for (Iterator<Map.Entry<String, Long>> i = index.iterator(tx); i.hasNext(); ) {
+            Map.Entry<String, Long> entry = (Map.Entry<String, Long>) i.next();
+            assertEquals(key(counter), entry.getKey());
+            assertEquals(counter, (long) entry.getValue());
+            counter++;
+        }
+        assertEquals("We iterated over all entries", entryCount, counter);
+
+        this.index.unload(tx);
+        tx.commit();
+    }
+
+    public void testVisitor() throws Exception {
+        createPageFileAndIndex(100);
+        ListIndex<String, Long> index = ((ListIndex<String, Long>) this.index);
+        this.index.load(tx);
+        tx.commit();
+
+        // Insert in reverse order..
+        doInsert(1000);
+
+        this.index.unload(tx);
+        tx.commit();
+        this.index.load(tx);
+        tx.commit();
+
+        // BTree should iterate it in sorted order.
+
+        /*index.visit(tx, new BTreeVisitor<String, Long>(){
+            public boolean isInterestedInKeysBetween(String first, String second) {
+                return true;
+            }
+            public void visit(List<String> keys, List<Long> values) {
+            }
+        });*/
+
+
+        this.index.unload(tx);
+        tx.commit();
+    }
+
+
+    public void testRandomRemove() throws Exception {
+
+        createPageFileAndIndex(100);
+        ListIndex<String, Long> index = ((ListIndex<String, Long>) this.index);
+        this.index.load(tx);
+        tx.commit();
+
+        final int count = 4000;
+        doInsert(count);
+
+        Random rand = new Random(System.currentTimeMillis());
+        int i = 0, prev = 0;
+        while (!index.isEmpty(tx)) {
+            prev = i;
+            i = rand.nextInt(count);
+            try {
+                index.remove(tx, key(i));
+            } catch (Exception e) {
+                e.printStackTrace();
+                fail("unexpected exception on " + i + ", prev: " + prev + ", ex: " + e);
+            }
+        }
+    }
+
+    public void testRemovePattern() throws Exception {
+        createPageFileAndIndex(100);
+        ListIndex<String, Long> index = ((ListIndex<String, Long>) this.index);
+        this.index.load(tx);
+        tx.commit();
+
+        final int count = 4000;
+        doInsert(count);
+
+        index.remove(tx, key(3697));
+        index.remove(tx, key(1566));
+    }
+
+    public void testLargeAppendTimed() throws Exception {
+        createPageFileAndIndex(100);
+        ListIndex<String, Long> listIndex = ((ListIndex<String, Long>) this.index);
+        this.index.load(tx);
+        tx.commit();
+        final int COUNT = 50000;
+        long start = System.currentTimeMillis();
+        for (int i = 0; i < COUNT; i++) {
+            //String test = new String("test" + i);
+            //ByteSequence bs = new ByteSequence(test.getBytes());
+             listIndex.put(tx, key(i), (long) i);
+             tx.commit();
+        }
+        LOG.info("Time to add " + COUNT + ": " + (System.currentTimeMillis() - start) + " mills");
+        LOG.info("Page count: " + listIndex.getPageFile().getPageCount());
+    }
+
+    void doInsertReverse(int count) throws Exception {
+        for (int i = count - 1; i >= 0; i--) {
+            ((ListIndex) index).addFirst(tx, key(i), (long) i);
+            tx.commit();
+        }
+    }
+
+    @Override
+    protected String key(int i) {
+        return "key:" + nf.format(i);
+    }
+}