Benchmark CouchDB reduce view with buit-in reducer

reduce_group_bench.py

#!/usr/bin/env python3
#
# Benchmark built-in reduce
# Example:
#   reduce_group_bench.py --ndocs 200000

import time
import argparse
import requests
import statistics

URL = "http://adm:pass@127.0.0.1:15984"
DB = "reduce_group_bench"
DDOC = "bench"
VIEW = "v"
TIMEOUT = 600

def doc_key(j, args):
    # every doc gets a distinct key
    # group_level=0 => string key
    # otherwise use three level array key for use with groups=1,2,3
    if args.group_level == 0:
        return f"{j:020d}"
    a, rem = divmod(j, 10000)
    b, c = divmod(rem, 100)
    return [f"{a:06d}", f"{b:03d}", f"{c:03d}"]

def query_params(args):
    if args.group_level == 0:
        return {"group": "true"}
    return {"group_level": str(args.group_level)}

def main(args):
    s = requests.Session()
    s.auth = ("adm", "pass")
    base = args.url.rstrip("/")

    def get(path, **kw):
        r = s.get(f"{base}/{path}", timeout=TIMEOUT, **kw)
        r.raise_for_status()
        return r

    def put(path, **kw):
        r = s.put(f"{base}/{path}", timeout=TIMEOUT, **kw)
        r.raise_for_status()
        return r

    def post(path, **kw):
        r = s.post(f"{base}/{path}", timeout=TIMEOUT, **kw)
        r.raise_for_status()
        return r

    # new db
    if s.head(f"{base}/{args.db}", timeout=TIMEOUT).status_code == 200:
        s.delete(f"{base}/{args.db}", timeout=TIMEOUT)
    put(f"{args.db}?q={args.q}&n={args.n}")

    mode = "group=true" if args.group_level == 0 else f"group_level={args.group_level}"
    print(f"db={args.db} q={args.q} n={args.n} ndocs={args.ndocs} {mode}")

    print("loading...")
    t0 = time.monotonic()
    for i in range(0, args.ndocs, args.batch):
        end = min(i + args.batch, args.ndocs)
        docs = [{"k": doc_key(j, args)} for j in range(i, end)]
        post(f"{args.db}/_bulk_docs", json={"docs": docs})
    print(f"  loaded {args.ndocs} docs in {time.monotonic()-t0:.1f}s")

    view = {"map": "function(doc){emit(doc.k, 1);}", "reduce": "_count"}
    put(f"{args.db}/_design/{DDOC}", json={"views": {VIEW: view}, "autoupdate": False})

    view_path = f"{args.db}/_design/{DDOC}/_view/{VIEW}"
    print("building index...")
    t0 = time.monotonic()
    get(view_path, params={**query_params(args), "limit": 1, "stale": "update_after"})
    deadline = time.monotonic() + TIMEOUT
    while time.monotonic() < deadline:
        info = get(f"{args.db}/_design/{DDOC}/_info").json()["view_index"]
        if info.get("updater_running") is False and info.get("update_seq", 0) >= args.ndocs:
            break
        time.sleep(0.2)
    print(f"  built in {time.monotonic()-t0:.1f}s")

    # dt group=true reduce query
    print(f"timing {mode} query x{args.reps}...")
    samples = []
    for _ in range(args.reps):
        t0 = time.monotonic()
        r = get(view_path, params=query_params(args))
        dt = (time.monotonic() - t0) * 1000
        nrows = len(r.json()["rows"])
        samples.append(dt)
    samples.sort()
    print(f"  rows={nrows}")
    print(
        f"  min={samples[0]:.1f}ms  median={statistics.median(samples):.1f}ms  "
        f"max={samples[-1]:.1f}ms"
    )

if __name__ == "__main__":
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--url", default=URL)
    p.add_argument("--db", default=DB)
    p.add_argument("--ndocs", type=int, default=200000)
    p.add_argument("--group-level", type=int, default=0, help="0 = strings, 1,2,3 = array levels")
    p.add_argument("--q", type=int, default=8, help="Db Q")
    p.add_argument("--n", type=int, default=1, help="Db N")
    p.add_argument("--batch", type=int, default=2000)
    p.add_argument("--reps", type=int, default=7)
    main(p.parse_args())

./reduce_group_bench.py --ndocs 100000 --q 8 --reps 7
db=reduce_group_bench q=8 n=1 ndocs=100000 group=true
loading...
  loaded 100000 docs in 4.4s
building index...
  built in 4.5s
timing group=true query x7...
  rows=100000
  min=5644.0ms  median=5786.6ms  max=6108.0ms