def test_bulk_load_col_delete(self):
        if self.keyfmt != 'r':
                return

        uri = self.type + self.name
        self.session.create(uri,
            'key_format=' + self.keyfmt + ',value_format=' + self.valfmt)
        cursor = self.session.open_cursor(uri, None, "bulk")
        for i in range(1, 1000):
            if i % 7 == 0:
                cursor[simple_key(cursor, i)] = simple_value(cursor, i)

        # Ensure we create all the missing records.
        i = i + 1
        cursor[simple_key(cursor, i)] = simple_value(cursor, i)

        cursor.close()
        cursor = self.session.open_cursor(uri, None, None)

        # Verify all the records are there, in their proper state.
        for i in range(1, 1000):
            cursor.set_key(simple_key(cursor, i))
            if i % 7 == 0:
                cursor.search()
                self.assertEqual(cursor.get_value(), simple_value(cursor, i))
            elif cursor.value_format == '8t':
                cursor.search()
                self.assertEqual(cursor.get_value(), 0)
            else:
                self.assertEqual(cursor.search(), wiredtiger.WT_NOTFOUND)

 def test_truncate_empty_cursor(self):
     uri = self.type + self.name
     self.session.create(uri,
         ',key_format=' + self.keyfmt + ',value_format=S')
     c1 = self.session.open_cursor(uri, None)
     c1.set_key(simple_key(c1, 1000))
     c2 = self.session.open_cursor(uri, None)
     c2.set_key(simple_key(c2, 2000))
     self.assertEqual(self.session.truncate(None, c1, c2, None), 0)

    def test_colmax_op(self):
        recno = self.recno

        uri = self.type + self.name
        self.session.create(uri, 'key_format=r' +',value_format=' + self.valfmt)

        # Insert a big record with/without a bulk cursor.
        bulk_config = ""
        if self.bulk:
            bulk_config = "bulk"
        cursor = self.session.open_cursor(uri, None, bulk_config)

        # Optionaly make the big record the only record in the table.
        if not self.single:
            for i in range(1, 723):
                cursor[simple_key(cursor, i)] = simple_value(cursor, i)

        # Confirm searching past the end of the table works.
        if not self.bulk:
            cursor.set_key(recno)
            self.assertEqual(cursor.search(), wiredtiger.WT_NOTFOUND)

        # Insert the big record.
        cursor[simple_key(cursor, recno)] = simple_value(cursor, recno)

        # Optionally flush to disk; re-open the cursor as necessary.
        if self.bulk or self.reopen:
            cursor.close()
        if self.reopen == 1:
            self.reopen_conn()
        if self.bulk or self.reopen:
            cursor = self.session.open_cursor(uri, None, None)

        # Search for the large record.
        cursor.set_key(recno)
        self.assertEqual(cursor.search(), 0)
        self.assertEqual(cursor.get_value(), simple_value(cursor, recno))

        # Update it.
        cursor[simple_key(cursor, recno)] = simple_value(cursor, 37)
        cursor.set_key(recno)
        self.assertEqual(cursor.search(), 0)
        self.assertEqual(cursor.get_value(), simple_value(cursor, 37))

        # Remove it.
        cursor.set_key(recno)
        self.assertEqual(cursor.remove(), 0)
        cursor.set_key(simple_key(cursor, recno))
        self.assertEqual(cursor.search(), wiredtiger.WT_NOTFOUND)

    def test_bulk_load_order_check(self):
        uri = self.type + self.name
        self.session.create(uri,
            'key_format=' + self.keyfmt + ',value_format=' + self.valfmt)
        cursor = self.session.open_cursor(uri, None, "bulk")
        cursor[simple_key(cursor, 10)] = simple_value(cursor, 10)

        for i in [1, 9, 10]:
            cursor.set_key(simple_key(cursor, 1))
            cursor.set_value(simple_value(cursor, 1))
            msg = '/than previously inserted key/'
            self.assertRaisesWithMessage(
                wiredtiger.WiredTigerError, lambda: cursor.insert(), msg)

        cursor[simple_key(cursor, 11)] = simple_value(cursor, 11)

    def empty(self):
        uri = self.type + self.name
        self.session = self.conn.open_session()
        self.session.create(uri, 'key_format=' + self.fmt)

        # Add a few records to the object and remove them.
        cursor = self.session.open_cursor(uri, None, None)
        for i in range(1,5):
            key = simple_key(cursor, i)
            cursor[key] = "XXX"
            del cursor[key]

        # Perform a checkpoint (we shouldn't write any underlying pages because
        # of a checkpoint, either).
        self.session.checkpoint("name=ckpt")

        # Open and close a checkpoint cursor.
        cursor = self.session.open_cursor(uri, None, "checkpoint=ckpt")
        cursor.close()

        self.session.close()

        # The file should not have grown.
        name = self.name
        if self.type == "table:":
            name = name + '.wt'
        self.assertEquals(os.stat(name).st_size, 4*1024)

 def test_bulk_load(self):
     uri = self.type + self.name
     self.session.create(uri,
         'key_format=' + self.keyfmt + ',value_format=' + self.valfmt)
     cursor = self.session.open_cursor(uri, None, "bulk")
     for i in range(1, 1000):
         cursor[simple_key(cursor, i)] = simple_value(cursor, i)

    def test_bulk_load_var_append(self):
        if self.keyfmt != 'r':
                return

        uri = self.type + self.name
        self.session.create(uri,
            'key_format=' + self.keyfmt + ',value_format=' + self.valfmt)
        cursor = self.session.open_cursor(uri, None, "bulk,append")
        for i in range(1, 1000):
            cursor[simple_key(cursor, 37)] = simple_value(cursor, i)
        cursor.close()
        cursor = self.session.open_cursor(uri, None, None)
        for i in range(1, 1000):
            cursor.set_key(simple_key(cursor, i))
            cursor.search()
            self.assertEqual(cursor.get_value(), simple_value(cursor, i))

    def test_bug006(self):
        uri = self.uri + self.name
        self.session.create(uri, 'value_format=S,key_format=S')
        cursor = self.session.open_cursor(uri, None)
        for i in range(1, 1000):
            cursor[simple_key(cursor, i)] = simple_value(cursor, i)

        # Table operations should fail, the cursor is open.
        self.assertRaises(
            wiredtiger.WiredTigerError, lambda: self.session.drop(uri, None))
        self.assertRaises(
            wiredtiger.WiredTigerError,
            lambda: self.session.rename(uri, self.uri + "new", None))
        self.assertRaises(
            wiredtiger.WiredTigerError, lambda: self.session.salvage(uri, None))
        self.assertRaises(
            wiredtiger.WiredTigerError, lambda: self.session.upgrade(uri, None))
        self.assertRaises(
            wiredtiger.WiredTigerError, lambda: self.session.verify(uri, None))

        cursor.close()

        # Table operations should succeed, the cursor is closed.
        self.session.rename(uri, self.uri + "new", None)
        self.session.rename(self.uri + "new", uri, None)
        self.session.salvage(uri, None)
        self.session.truncate(uri, None, None, None)
        self.session.upgrade(uri, None)
        self.session.verify(uri, None)

        self.session.drop(uri, None)

 def test_bulk_load_busy(self):
     uri = self.type + self.name
     self.session.create(uri, 'key_format=S,value_format=S')
     cursor = self.session.open_cursor(uri, None)
     cursor[simple_key(cursor, 1)] = simple_value(cursor, 1)
     # Don't close the insert cursor, we want EBUSY.
     self.assertRaises(wiredtiger.WiredTigerError,
         lambda: self.session.open_cursor(uri, None, "bulk"))

    def test_bug004(self):
        # Create the object, fill with overflow keys and values.
        self.session.create(self.uri, self.config)
        c1 = self.session.open_cursor(self.uri, None)
        for i in range(1, self.nentries):
            c1[simple_key(c1, i) + 'abcdef' * 100] = \
                simple_value(c1, i) + 'abcdef' * 100
        c1.close()

        # Verify the object, force it to disk, and verify the on-disk version.
        self.session.verify(self.uri)
        self.reopen_conn()
        self.session.verify(self.uri)

        # Create a new session and start a transaction to force the engine
        # to access old versions of the key/value pairs.
        tmp_session = self.conn.open_session(None)
        tmp_session.begin_transaction("isolation=snapshot")

        # Load the object and use truncate to delete a set of records.  (I'm
        # using truncate because it doesn't instantiate keys, all other ops
        # currently do -- that's unlikely to change, but is a problem for the
        # test going forward.)
        c1 = self.session.open_cursor(self.uri, None)
        c1.set_key(simple_key(c1, self.nentries - 5) + 'abcdef' * 100)
        c2 = self.session.open_cursor(self.uri, None)
        c2.set_key(simple_key(c2, self.nentries + 5) + 'abcdef' * 100)
        self.session.truncate(None, c1, c2, None)
        c1.close()
        c2.close()

        # Checkpoint, freeing overflow blocks.
        self.session.checkpoint()

        # Use the snapshot cursor to retrieve the old key/value pairs
        c1 = tmp_session.open_cursor(self.uri, None)
        c1.set_key(simple_key(c1, 1) + 'abcdef' * 100)
        c1.search()
        for i in range(2, self.nentries):
            c1.next()
            self.assertEquals(
                c1.get_key(), simple_key(c1, i) + 'abcdef' * 100)
            self.assertEquals(
                c1.get_value(), simple_value(c1, i) + 'abcdef' * 100)

    def test_cursor_random_invisible_before(self):
        uri = self.type
        self.session.create(uri, 'key_format=S,value_format=S')
        cursor = self.session.open_cursor(uri, None)

        # Insert a single leading record.
        cursor[simple_key(cursor, 99)] = simple_value(cursor, 99)

        # Start a transaction.
        self.session.begin_transaction()
        for i in range(2, 100):
            cursor[simple_key(cursor, i)] = simple_value(cursor, i)

        # Open another session, the updates won't yet be visible, we should
        # return the only possible record.
        s = self.conn.open_session()
        cursor = s.open_cursor(uri, None, self.config)
        self.assertEquals(cursor.next(), 0)
        self.assertEqual(cursor.get_key(), simple_key(cursor, 99))

 def test_bulk_load_not_empty(self):
     uri = self.type + self.name
     self.session.create(uri, 'key_format=S,value_format=S')
     cursor = self.session.open_cursor(uri, None)
     cursor[simple_key(cursor, 1)] = simple_value(cursor, 1)
     # Close the insert cursor, else we'll get EBUSY.
     cursor.close()
     msg = '/bulk-load is only supported on newly created objects/'
     self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
         lambda: self.session.open_cursor(uri, None, "bulk"), msg)

    def test_bulk_load_col_big(self):
        if self.keyfmt != 'r' or self.valfmt == '8t':
                return

        uri = self.type + self.name
        self.session.create(uri,
            'key_format=' + self.keyfmt + ',value_format=' + self.valfmt)
        cursor = self.session.open_cursor(uri, None, "bulk")
        for i in range(1, 10):
            cursor[simple_key(cursor, i)] = simple_value(cursor, i)

        # A big record -- if it's not efficient, we'll just hang.
        big = 18446744073709551606
        cursor[simple_key(cursor, big)] = simple_value(cursor, big)

        cursor.close()
        cursor = self.session.open_cursor(uri, None, None)
        cursor.set_key(simple_key(cursor, big))
        cursor.search()
        self.assertEqual(cursor.get_value(), simple_value(cursor, big))

    def test_dictionary(self):
        nentries = 25000
        uri = 'file:test_dictionary'    # This is a btree layer test.

        # Create the object, open the cursor, insert some records with identical values. Use
        # alternating values, otherwise column-store will RLE compress them into a single item.
        self.session.create(uri, 'dictionary=100,value_format=S,key_format=' + self.key_format)
        cursor = self.session.open_cursor(uri, None)
        i = 0
        while i < nentries:
            i = i + 1
            cursor[simple_key(cursor, i)] = "the same value as the odd items"
            i = i + 1
            cursor[simple_key(cursor, i)] = "the same value as the even items"
        cursor.close()

        # Checkpoint to force the pages through reconciliation.
        self.session.checkpoint()

        # Confirm the dictionary was effective.
        cursor = self.session.open_cursor('statistics:' + uri, None, None)
        self.assertGreater(cursor[stat.dsrc.rec_dictionary][2], nentries - 100)

    def test_bulk_load_row_order_nocheck(self):
        # Row-store offers an optional fast-past that skips the relatively
        # expensive key-order checks, used when the input is known to be
        # correct. Column-store comparisons are cheap, so it doesn't have
        # that fast-path support.
        if self.keyfmt != 'S':
                return

        uri = self.type + self.name
        self.session.create(uri,
            'key_format=' + self.keyfmt + ',value_format=' + self.valfmt)
        cursor = self.session.open_cursor(uri, None, "bulk,skip_sort_check")
        cursor[simple_key(cursor, 10)] = simple_value(cursor, 10)
        cursor[simple_key(cursor, 1)] = simple_value(cursor, 1)

        if not wiredtiger.diagnostic_build():
            self.skipTest('requires a diagnostic build')

        # Close explicitly, there's going to be a fallure.
        msg = '/are incorrectly sorted/'
        self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
            lambda: self.conn.close(), msg)

    def test_bulk_load_var_rle(self):
        if self.keyfmt != 'r' or self.valfmt == '8t':
                return

        # We can't directly test RLE, it's internal to WiredTiger. However,
        # diagnostic builds catch records that should have been RLE compressed,
        # but weren't, so setting matching values should be sufficient.
        uri = self.type + self.name
        self.session.create(uri,
            'key_format=' + self.keyfmt + ',value_format=' + self.valfmt)
        cursor = self.session.open_cursor(uri, None, "bulk")
        for i in range(1, 1000):
            cursor[simple_key(cursor, i)] = simple_value(cursor, i/7)

    def test_cursor_random_invisible_all(self):
        uri = self.type
        self.session.create(uri, 'key_format=S,value_format=S')
        cursor = self.session.open_cursor(uri, None)

        # Start a transaction.
        self.session.begin_transaction()
        for i in range(1, 100):
            cursor[simple_key(cursor, i)] = simple_value(cursor, i)

        # Open another session, the updates won't yet be visible, we shouldn't
        # find anything at all.
        s = self.conn.open_session()
        cursor = s.open_cursor(uri, None, self.config)
        self.assertEqual(cursor.next(), wiredtiger.WT_NOTFOUND)

    def test_row_store_empty_values(self):
        nentries = 25000
        uri = 'file:test_empty_values'          # This is a btree layer test.

        # Create the object, open the cursor, insert some records with zero-length values.
        self.session.create(uri, 'value_format=u,key_format=S')
        cursor = self.session.open_cursor(uri, None)
        for i in xrange(1, nentries + 1):
            cursor[simple_key(cursor, i)] = ""
        cursor.close()

        # Reopen to force the object to disk.
        self.reopen_conn()

        # Confirm the values weren't stored..
        cursor = self.session.open_cursor('statistics:' + uri, None, 'statistics=(tree_walk)')
        self.assertEqual(cursor[stat.dsrc.btree_row_empty_values][2], nentries)

    def test_bug005(self):
        # Create the object.
        self.session.create(self.uri, 'value_format=S,key_format=S')
        cursor = self.session.open_cursor(self.uri, None)
        for i in range(1, 1000):
            cursor[simple_key(cursor, i)] = simple_value(cursor, i)
        cursor.close()

        # Verify the object, force it to disk, and verify the on-disk version.
        self.session.verify(self.uri)
        self.reopen_conn()
        self.session.verify(self.uri)

        # Append random data to the end.
        f = open('test_bug005', 'a')
        f.write('random data')
        f.close()

        # Verify the object again.
        self.session.verify(self.uri)

    def test_bulk_backup(self):
        # Open a bulk cursor and insert a few records.
        self.session.create(self.uri, 'key_format=S,value_format=S')
        cursor = self.session.open_cursor(self.uri, None, 'bulk')
        for i in range(1, 10):
            cursor[simple_key(cursor, i)] = simple_value(cursor, i)

        # Test without a checkpoint, with an unnamed checkpoint, with a named
        # checkpoint.
        if self.ckpt_type == 'named':
            self.session.checkpoint('name=myckpt')
        elif self.ckpt_type == 'unnamed':
            self.session.checkpoint()

        # Test with the same and different sessions than the bulk-get call,
        # test both the database handle and session handle caches.
        if self.session_type == 'same':
            self.check_backup(self.session)
        else:
            self.check_backup(self.conn.open_session())