Merge lp:~pbeaman/akiban-persistit/bigload into lp:akiban-persistit

[Peter Beaman]

Add a new stress test that sorts random keys into trees in a temporary volume and then inserts them in key-sort order into the main database. This is primarily a demonstration developed to help an answer from a Persistit evaluator, but since it exercises temporary volumes and potential very large databases it seems appropriate to have the code available in the stress test suite.

This branch also adds a couple of new methods to the Key class. The original version of BigLoad used these. We also have another potential user who may benefit from these and they have some general usefulness so are therefore proposed. New unit tests cover them.

[Peter Beaman (pbeaman)]

Note: removed a change in the Exchange class subsequent to original merge proposal. Exchange is now unchanged.

[Peter Beaman (pbeaman)]

Note that appendKeySegment does not check the size of the destination key - an AIOOBE can result. However, that's also true of the other append methods, so this new method behaves no worse.

[Nathan Williams (nwilliams)]

New test sounds good, though I didn't inspect it too closely.

Mainline additions look good too.

review: Approve

[Akiban Build User (build-akiban)]

There were 2 failures during build/test:

* job persistit-build failed at build number 470: http://172.16.20.104:8080/job/persistit-build/470/

* view must-pass failed: persistit-build is yellow

review: Needs Fixing

[Peter Beaman (pbeaman)]

Known sporadic failure in TransactionLifetimeTest. I'm re-Approving this since the changes in this proposal didn't go anywhere near the TLT failure.

=== modified file 'src/main/java/com/persistit/BufferPool.java'
--- src/main/java/com/persistit/BufferPool.java	2012-10-05 14:21:47 +0000
+++ src/main/java/com/persistit/BufferPool.java	2012-11-01 20:16:23 +0000
@@ -653,7 +653,10 @@
             }
         }
         buffer.clearValid();
-        buffer.clearDirty();
+        if (buffer.isDirty()) {
+            _dirtyPageCount.decrementAndGet();
+            buffer.clearDirty();
+        }
         buffer.setPageAddressAndVolume(0, null);
     }
 

=== modified file 'src/main/java/com/persistit/Key.java'
--- src/main/java/com/persistit/Key.java	2012-08-24 13:57:19 +0000
+++ src/main/java/com/persistit/Key.java	2012-11-01 20:16:23 +0000
@@ -1146,7 +1146,7 @@
     }
 
     /**
-     * Compares a bounded subarray of the backing byte array for this
+     * Compare a bounded subarray of the backing byte array for this
      * <code>Key</code> with another <code>Key</code>. This method is intended
      * for use within the library and is generally not useful for applications.
      * 
@@ -1162,8 +1162,9 @@
      *         to the corresponding fragment of the supplied <code>Key</code>.
      */
     public int compareKeyFragment(final Key key, final int fragmentStart, final int fragmentSize) {
-        if (key == this)
+        if (key == this) {
             return 0;
+        }
         final int size1 = this._size;
         final int size2 = key.getEncodedSize();
         final byte[] bytes1 = this.getEncodedBytes();
@@ -1178,8 +1179,9 @@
         for (int i = fragmentStart; i < size; i++) {
             final int b1 = bytes1[i] & 0xFF;
             final int b2 = bytes2[i] & 0xFF;
-            if (b1 != b2)
+            if (b1 != b2) {
                 return b1 - b2;
+            }
         }
         if (size == fragmentSize + fragmentStart)
             return 0;
@@ -1191,6 +1193,42 @@
     }
 
     /**
+     * Compare the next key segment of this key to the next key segment of the
+     * supplied Key. The next key segment is determined by the current index of
+     * the key and can be set using the {@link #setIndex(int)} method. Returns a
+     * positive integer if the next segment of this <code>Key</code> is larger
+     * than the next segment of the supplied <code>Key</code>, a negative
+     * integer if it is smaller, or zero if the segments are equal.
+     * 
+     * @param key
+     * @return the comparison result
+     */
+    public int compareKeySegment(final Key key) {
+        if (key == this) {
+            return 0;
+        }
+        final byte[] bytes1 = this.getEncodedBytes();
+        final byte[] bytes2 = key.getEncodedBytes();
+        final int index1 = this.getIndex();
+        final int index2 = key.getIndex();
+        final int count1 = this.getEncodedSize() - this.getIndex();
+        final int count2 = key.getEncodedSize() - key.getIndex();
+        final int count = Math.min(count1, count2);
+
+        for (int i = 0; i < count; i++) {
+            final int b1 = bytes1[i + index1] & 0xFF;
+            final int b2 = bytes2[i + index2] & 0xFF;
+            if (b1 != b2) {
+                return b1 - b2;
+            }
+            if (b1 == 0) {
+                return 0;
+            }
+        }
+        return count1 - count2;
+    }
+
+    /**
      * Returns the index of the first encoded byte of this key that is different
      * than the corresponding byte of the supplied <code>Key</code>. If all
      * bytes match, then returns the encoded length of the shorter key.
@@ -2037,6 +2075,25 @@
     }
 
     /**
+     * Append the next key segment of the supplied <code>Key</code> to this
+     * <code>Key</code. The next key segment is determined by the current index
+     * of the key and can be set using the {@link #setIndex(int)} method.
+     * 
+     * @param key
+     */
+    public void appendKeySegment(final Key key) {
+        int length = 0;
+        for (int index = key.getIndex(); index < key.getEncodedSize(); index++) {
+            length++;
+            if (key.getEncodedBytes()[index] == 0) {
+                break;
+            }
+        }
+        System.arraycopy(key.getEncodedBytes(), key.getIndex(), _bytes, _size, length);
+        _size += length;
+    }
+
+    /**
      * Replaces the final key segment with the supplied boolean value. If the
      * key is currently empty, this method simply appends the value. This method
      * is equivalent to invoking {@link #cut} and then {@link #append(boolean)}.

=== added file 'src/test/java/com/persistit/stress/InsertBigLoad.java'
--- src/test/java/com/persistit/stress/InsertBigLoad.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/persistit/stress/InsertBigLoad.java	2012-11-01 20:16:23 +0000
@@ -0,0 +1,55 @@
+/**
+ * Copyright © 2012 Akiban Technologies, Inc.  All rights reserved.
+ * 
+ * This program and the accompanying materials are made available
+ * under the terms of the Eclipse Public License v1.0 which
+ * accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ * 
+ * This program may also be available under different license terms.
+ * For more information, see www.akiban.com or contact [e-mail].
+ * 
+ * Contributors:
+ * Akiban Technologies, Inc.
+ */
+
+package com.persistit.stress;
+
+import com.persistit.Configuration;
+import com.persistit.Persistit;
+import com.persistit.Transaction.CommitPolicy;
+import com.persistit.stress.unit.BigLoad;
+
+public class InsertBigLoad extends AbstractSuite {
+
+    static String name() {
+        return InsertBigLoad.class.getSimpleName();
+    }
+
+    public static void main(final String[] args) throws Exception {
+        new InsertBigLoad(args).runTest();
+    }
+
+    public InsertBigLoad(final String[] args) {
+        super(name(), args);
+    }
+
+    @Override
+    public void runTest() throws Exception {
+
+        deleteFiles(substitute("$datapath$/persistit*"));
+
+        add(new BigLoad("records=10000000 buckets=10"));
+
+        final Configuration config = makeConfiguration(16384, "50000", CommitPolicy.SOFT);
+        config.setTmpVolMaxSize(100000000000l);
+        final Persistit persistit = new Persistit();
+        persistit.initialize(config);
+
+        try {
+            execute(persistit);
+        } finally {
+            persistit.close();
+        }
+    }
+}

=== added file 'src/test/java/com/persistit/stress/unit/BigLoad.java'
--- src/test/java/com/persistit/stress/unit/BigLoad.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/persistit/stress/unit/BigLoad.java	2012-11-01 20:16:23 +0000
@@ -0,0 +1,243 @@
+/**
+ * Copyright © 2012 Akiban Technologies, Inc.  All rights reserved.
+ * 
+ * This program and the accompanying materials are made available
+ * under the terms of the Eclipse Public License v1.0 which
+ * accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ * 
+ * This program may also be available under different license terms.
+ * For more information, see www.akiban.com or contact [e-mail].
+ * 
+ * Contributors:
+ * Akiban Technologies, Inc.
+ */
+
+package com.persistit.stress.unit;
+
+import java.util.Random;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+import com.persistit.Exchange;
+import com.persistit.Key;
+import com.persistit.Persistit;
+import com.persistit.Volume;
+import com.persistit.exception.PersistitException;
+import com.persistit.stress.AbstractStressTest;
+import com.persistit.util.ArgParser;
+import com.persistit.util.Util;
+
+/**
+ * <p>
+ * Simulate loading a large set (e.g., 500M) of records with large random keys.
+ * Because straightforward insertion results in highly randomized page access
+ * after the database size has exceed the amount of buffer pool memory space,
+ * this demo creates smaller sorted sets of keys and then merges them to create
+ * the final Tree in sequential order. As a side-effect, the final tree is also
+ * physically coherent in that the logical and physical order of keys disk are
+ * closely aligned.
+ * </p>
+ * <p>
+ * This class can be run stand-alone through its static main method, or within
+ * the stress test suite.
+ * </p>
+ * 
+ * @author peter
+ */
+public class BigLoad extends AbstractStressTest {
+
+    private static final Random RANDOM = new Random();
+
+    private int totalRecords;
+    private int recordsPerBucket;
+
+    /**
+     * A Comparable wrapper for an Exchange. An instance of this class may be
+     * held in a SortedMap only if the Key of the Exchange does not change. In
+     * this example, the ComparableExchangeHolder is always removed from the
+     * TreeMap before the Key changes and then reinserted into a new location
+     * after the key has changed.
+     */
+    static class ComparableExchangeHolder implements Comparable<ComparableExchangeHolder> {
+
+        final Exchange exchange;
+
+        ComparableExchangeHolder(final Exchange exchange) {
+            this.exchange = exchange;
+        }
+
+        @Override
+        public int compareTo(final ComparableExchangeHolder ceh) {
+            final Key k1 = exchange.getKey();
+            final Key k2 = ceh.exchange.getKey();
+            return k1.compareTo(k2);
+        }
+    }
+
+    public BigLoad(final int totalRecords, final int buckets) {
+        super("");
+        this.totalRecords = totalRecords;
+        this.recordsPerBucket = totalRecords / buckets;
+    }
+
+    public void load(final Persistit db) throws PersistitException {
+        final long startLoadTime = System.nanoTime();
+        final Volume sortVolume = db.createTemporaryVolume();
+        final Exchange resultExchange = db.getExchange("persistit", "sorted", true);
+        System.out.printf("Loading %,d records into %,d buckets\n", totalRecords, totalRecords / recordsPerBucket);
+        final int bucketCount = loadBuckets(db, sortVolume);
+        final long endLoadTime = System.nanoTime();
+
+        System.out.printf("Merging %,d records from %,d buckets into main database\n", totalRecords, bucketCount);
+        mergeBuckets(db, bucketCount, sortVolume, resultExchange);
+        final long endMergeTime = System.nanoTime();
+        System.out.printf("Merged %,d records in %,dms\n", totalRecords, (endMergeTime - endLoadTime) / Util.NS_PER_MS);
+        sortVolume.close();
+        System.out.printf("Counting keys in main database (100M keys per dot) ");
+        resultExchange.clear().append(Key.BEFORE);
+        long count = 0;
+        while (resultExchange.next()) {
+            count++;
+            if ((count % 100000000) == 0) {
+                System.out.print(".");
+                System.out.flush();
+            }
+        }
+        final long endCountTime = System.nanoTime();
+        System.out.printf("\nCounted %,d keys in the main database in %,dms\n", count, (endCountTime - endMergeTime)
+                / Util.NS_PER_MS);
+        System.out.printf("Total time to load, merge and count %,d records is %,dms", totalRecords,
+                (endCountTime - startLoadTime) / Util.NS_PER_MS);
+    }
+
+    private int loadBuckets(final Persistit db, final Volume volume) throws PersistitException {
+        long bucketStartTime = 0;
+        int bucket = 0;
+        Exchange ex = null;
+        for (int i = 0; i < totalRecords; i++) {
+            if ((i % recordsPerBucket) == 0) {
+                final long now = System.nanoTime();
+                if (i > 0) {
+                    System.out.printf("Loaded bucket %,5d in %,12dms\n", bucket, (now - bucketStartTime)
+                            / Util.NS_PER_MS);
+                }
+                bucketStartTime = now;
+                bucket++;
+                ex = db.getExchange(volume, "sort" + bucket, true);
+            }
+            ex.clear().append(randomKey());
+            ex.store();
+        }
+        return bucket;
+    }
+
+    private void mergeBuckets(final Persistit db, final int bucketCount, final Volume sortVolume, final Exchange to)
+            throws PersistitException {
+        final long startLoadTime = System.nanoTime();
+        int loaded = 0;
+
+        final SortedMap<ComparableExchangeHolder, Integer> sortMap = new TreeMap<ComparableExchangeHolder, Integer>();
+        /*
+         * Load the sortMap using as keys the first key of each bucket.
+         */
+        for (int bucket = 1; bucket <= bucketCount; bucket++) {
+            final Exchange ex = db.getExchange(sortVolume, "sort" + bucket, false);
+            if (ex.append(Key.BEFORE).next()) {
+                final Integer duplicate = sortMap.put(new ComparableExchangeHolder(ex), bucket);
+                showDuplicate(duplicate, bucket, ex);
+            }
+        }
+
+        while (!sortMap.isEmpty()) {
+            final ComparableExchangeHolder ceh = sortMap.firstKey();
+            final int bucket = sortMap.remove(ceh);
+            ceh.exchange.getKey().copyTo(to.getKey());
+            if (ceh.exchange.next()) {
+                final Integer duplicate = sortMap.put(ceh, bucket);
+                showDuplicate(duplicate, bucket, ceh.exchange);
+            }
+            to.store();
+            if ((++loaded % 10000000) == 0) {
+                System.out.printf("Merged %,d records in %,dms\n", loaded, (System.nanoTime() - startLoadTime)
+                        / Util.NS_PER_MS);
+            }
+        }
+    }
+
+    private String randomKey() {
+        return String.format("%020dxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+                (RANDOM.nextLong() & Long.MAX_VALUE));
+    }
+
+    private void showDuplicate(final Integer bucket1, final int bucket2, final Exchange ex) {
+        if (bucket1 != null) {
+            System.out.printf("Duplicate key %s in buckets %,d and %,d\n", ex.getKey(), bucket1, bucket2);
+        }
+    }
+
+    /**
+     * Arguments:
+     * 
+     * records - total records to load
+     * 
+     * buckets - number of subdivisions to create while loading
+     * 
+     * propertiesPath - path to properties file for Persistit initialization
+     * 
+     * @param args
+     * @throws Exception
+     */
+
+    public static void main(final String[] args) throws Exception {
+        final int records = args.length > 0 ? Integer.parseInt(args[0]) : 1000000;
+        final int buckets = args.length > 1 ? Integer.parseInt(args[1]) : 100;
+        final BigLoad bl = new BigLoad(records, buckets);
+        final Persistit db = new Persistit();
+        if (args.length > 2) {
+            db.initialize(args[2]);
+        } else {
+            db.initialize();
+        }
+        try {
+            bl.load(db);
+        } finally {
+            db.close();
+        }
+    }
+
+    /*
+     * ----------------------------------------------------------------------
+     * 
+     * Stuff below this line is required to run within the stress test suite
+     * 
+     * ----------------------------------------------------------------------
+     */
+
+    public BigLoad(final String argsString) {
+        super(argsString);
+    }
+
+    private final static String[] ARGS_TEMPLATE = { "records|int:1000000:1:1000000000|Total records to create",
+            "buckets|int:100:1:1000000|Number of sort buckets", "tmpdir|string:|Temporary volume path" };
+
+    /**
+     * Method to parse stress test arguments passed by the stress test suite.
+     */
+    @Override
+    public void setUp() throws Exception {
+        super.setUp();
+        final ArgParser ap = new ArgParser("com.persistit.BigLoad", _args, ARGS_TEMPLATE).strict();
+        totalRecords = ap.getIntValue("records");
+        recordsPerBucket = totalRecords / ap.getIntValue("buckets");
+        final String path = ap.getStringValue("tmpdir");
+        if (path != null && !path.isEmpty()) {
+            getPersistit().getConfiguration().setTmpVolDir(path);
+        }
+    }
+
+    @Override
+    protected void executeTest() throws Exception {
+        load(getPersistit());
+    }
+}

=== modified file 'src/test/java/com/persistit/unit/KeyTest1.java'
--- src/test/java/com/persistit/unit/KeyTest1.java	2012-08-24 13:57:19 +0000
+++ src/test/java/com/persistit/unit/KeyTest1.java	2012-11-01 20:16:23 +0000
@@ -945,6 +945,49 @@
         }
     }
 
+    @Test
+    public void testCompareKeySegment() throws Exception {
+        final Key key1 = newKey();
+        final Key key2 = newKey();
+        key1.append(1).append(2).append(3).append("abc");
+        key2.append(3).append(2).append(1).append("abcd");
+        key1.indexTo(1);
+        key2.indexTo(1);
+        assertTrue("Should be == 0", 0 == key1.compareKeySegment(key2));
+        assertTrue("Should be == 0", 0 == key2.compareKeySegment(key1));
+        key1.indexTo(0);
+        assertTrue("Should be < 0", 0 > key1.compareKeySegment(key2));
+        assertTrue("Should be > 0", 0 < key2.compareKeySegment(key1));
+        key1.indexTo(2);
+        assertTrue("Should be > 0", 0 < key1.compareKeySegment(key2));
+        assertTrue("Should be < 0", 0 > key2.compareKeySegment(key1));
+        key1.indexTo(3);
+        assertTrue("Should be > 0", 0 < key1.compareKeySegment(key2));
+        assertTrue("Should be < 0", 0 > key2.compareKeySegment(key1));
+        key2.indexTo(3);
+        assertTrue("Should be < 0", 0 > key1.compareKeySegment(key2));
+        assertTrue("Should be > 0", 0 < key2.compareKeySegment(key1));
+    }
+
+    @Test
+    public void testAppendKeySegment() throws Exception {
+        final Key key1 = newKey();
+        final Key key2 = newKey();
+        key1.append(1).append(2).append(3).append("abc");
+        key1.indexTo(3);
+        key2.appendKeySegment(key1);
+        key1.indexTo(2);
+        key2.appendKeySegment(key1);
+        key1.indexTo(1);
+        key2.appendKeySegment(key1);
+        key1.indexTo(0);
+        key2.appendKeySegment(key1);
+        assertEquals("Key value incorrect", "{\"abc\",3,2,1}", key2.toString());
+        key1.indexTo(3);
+        key1.appendKeySegment(key1);
+        assertEquals("Key value incorrect", "{1,2,3,\"abc\",\"abc\"}", key1.toString());
+    }
+
     private static boolean doubleEquals(final double f1, final double f2) {
         if (Double.isNaN(f1)) {
             return Double.isNaN(f2);