compaction-test.py

#! /usr/bin/env python

import time
import uuid
import couchdbkit


s = couchdbkit.Server("http://127.0.0.1:15986")

ordered_doc_id = 0
def ordered_docs(n):
    global ordered_doc_id
    ret = [{"_id": "%032d" % (ordered_doc_id+i)} for i in range(n)]
    ordered_doc_id += n
    return ret


def random_docs(n):
    return [{"_id": uuid.uuid4().hex} for _ in range(n)]


def do_test(doc_maker):
    try:
        s.delete_db("compaction_test")
    except:
        pass
    d = s.create_db("compaction_test")

    print doc_maker.func_name
    for i in range(100):
        d.bulk_save(doc_maker(100))

    import pprint
    before = d.info()
    d.compact()
    while(d.info()["compact_running"]):
        time.sleep(1)
    after = d.info()

    perc = 100.0 * float(after["disk_size"]) / float(before["disk_size"])
    print "Disk Size: %d / %d = %0.2f%%" % (
            after["disk_size"], before["disk_size"], perc)


    perc = 100.0 * float(after["other"]["data_size"]) / \
                        float(before["other"]["data_size"])
    print "Disk Size: %d / %d = %0.2f%%" % (
            after["other"]["data_size"], before["other"]["data_size"], perc)


do_test(ordered_docs)
do_test(random_docs)