DynamoDB performance drop?

[javafanboy]

I think I am seeing a tendency of the curent DDB implementation in floci to have increasing latency (also for the basic operations that "should be O(1)" as the dataset increase in size - is this known or should I try confurming my suspicion with some reproducible test?

[hectorvent]

Hi @javafanboy,
I haven't notice it. I will investigate it deeper.

[Mohith1612]

@javafanboy your suspicion holds up. I measured it on floci:2.3.0 and on nightly-10252026, and two parts of DynamoDB get slower as a table grows: writes, and Query. GetItem doesn't.

Setup: one table with pk (S) and sk (S) keys, items of about 250 bytes spread over 1,000 partitions, loaded with BatchWriteItem. At each size I timed 300 sequential calls of each operation from boto3 on the same host. PutItem and UpdateItem overwrite a random existing item, so the table doesn't grow while I'm measuring. Query always reads the same 20-item partition, so the result size never changes. Times are p50 in ms.

FLOCI_STORAGE_MODE=memory (the default), 2.3.0

items	GetItem	PutItem	UpdateItem	Query (20 items)
1,000	0.95	0.98	1.05	1.32
10,000	1.00	1.97	1.64	4.19
50,000	0.97	8.34	8.14	18.04
100,000	0.95	15.95	16.14	33.07

The nightly build gives the same numbers, with either ITEM_LAYOUT.

FLOCI_STORAGE_MODE=persistent

items	PutItem 2.3.0 (single-file)	PutItem nightly (per-table, 16 shards)	Query (both)
1,000	3.41	3.14	~1.3
10,000	47.90	3.77	4–11
50,000	285.81	9.64	18–44

Loading 50k items took 268 s on 2.3.0 in persistent mode, compared with 7 s in memory.

Here's what's going on in the code (DynamoDbService.java at 51ce672):

• Query walks the whole table. Items are stored in a ConcurrentSkipListMap keyed pk#sk, but both query paths loop over items.values() and skip items whose partition key doesn't match (L1573, L4865). So a query costs O(table size), not O(partition size). On the base table it could read just that key's range of the map instead.
• Every write copies the table when the store can't write single items. persistItems does itemStore.put(storageKey, new HashMap<>(items)) (L414), and that path is also taken in memory mode. With the single-file layout in persistent mode, the whole JSON file gets rewritten on top of that, which is the 286 ms row.

What helps right now:

• If you're on persistent, the per-table layout from feat(dynamodb): opt-in per-table item files, split into shards #6303 (FLOCI_STORAGE_SERVICES_DYNAMODB_ITEM_LAYOUT=per-table, plus ..._ITEM_SHARDS set above 1) brings writes from 286 ms down to about 10 ms at 50k. It's on nightly, not in 2.3.0. hybrid or wal also take the file rewrite out of the request path.
• None of the settings changes Query. A large table is slow to query wherever it's stored.

The benchmark script, in case anyone wants to rerun it:

ddb_bench.py

"""Measure floci DynamoDB latency as a table grows.

usage: ddb_bench.py ENDPOINT LABEL SIZE[,SIZE...]
"""
import random
import statistics
import sys
import time
from concurrent.futures import ThreadPoolExecutor

import boto3
from botocore.config import Config

endpoint, label, sizes = sys.argv[1], sys.argv[2], [int(s) for s in sys.argv[3].split(",")]
OPS = 300
PAYLOAD = "x" * 200

cfg = Config(retries={"max_attempts": 10, "mode": "standard"}, max_pool_connections=16)
ddb = boto3.client("dynamodb", endpoint_url=endpoint, region_name="us-east-1",
                   aws_access_key_id="test", aws_secret_access_key="test", config=cfg)

TABLE = "bench"
ddb.create_table(TableName=TABLE, BillingMode="PAY_PER_REQUEST",
                 AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"},
                                       {"AttributeName": "sk", "AttributeType": "S"}],
                 KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"},
                            {"AttributeName": "sk", "KeyType": "RANGE"}])

# A fixed 20-item partition, so Query cost doesn't grow with the partition itself.
for i in range(20):
    ddb.put_item(TableName=TABLE, Item={"pk": {"S": "q"}, "sk": {"S": f"{i:04d}"}, "d": {"S": PAYLOAD}})


def key(i):
    return {"pk": {"S": f"p{i % 1000}"}, "sk": {"S": f"{i:08d}"}}


def write_batch(start, end):
    reqs = [{"PutRequest": {"Item": {**key(i), "d": {"S": PAYLOAD}}}} for i in range(start, end)]
    while reqs:
        resp = ddb.batch_write_item(RequestItems={TABLE: reqs})
        reqs = resp.get("UnprocessedItems", {}).get(TABLE, [])


def timed(fn):
    out = []
    for _ in range(OPS):
        t = time.perf_counter()
        fn()
        out.append((time.perf_counter() - t) * 1000)
    return out


def summary(ms):
    ms = sorted(ms)
    return f"{statistics.median(ms):7.2f} {ms[int(len(ms) * 0.95)]:7.2f}"


loaded = 0
print(f"# {label}")
print(f"{'items':>7} {'load_s':>7} | {'GetItem p50 p95':>15} | {'PutItem p50 p95':>15} | "
      f"{'UpdateItem p50 p95':>18} | {'Query p50 p95':>15}")
for size in sizes:
    t = time.perf_counter()
    with ThreadPoolExecutor(8) as pool:
        list(pool.map(lambda s: write_batch(s, min(s + 25, size)), range(loaded, size, 25)))
    load_s = time.perf_counter() - t
    loaded = size

    get = timed(lambda: ddb.get_item(TableName=TABLE, Key=key(random.randrange(size))))
    put = timed(lambda: ddb.put_item(TableName=TABLE,
                                     Item={**key(random.randrange(size)), "d": {"S": PAYLOAD}}))
    upd = timed(lambda: ddb.update_item(TableName=TABLE, Key=key(random.randrange(size)),
                                        UpdateExpression="SET n = if_not_exists(n, :z) + :o",
                                        ExpressionAttributeValues={":z": {"N": "0"}, ":o": {"N": "1"}}))
    qry = timed(lambda: ddb.query(TableName=TABLE, KeyConditionExpression="pk = :p",
                                  ExpressionAttributeValues={":p": {"S": "q"}}))
    print(f"{size:7d} {load_s:7.1f} | {summary(get):>15} | {summary(put):>15} | "
          f"{summary(upd):>18} | {summary(qry):>15}", flush=True)

[hectorvent]

Thanks a lot @Mohith1612 for putting this together. The tables, the code pointers and especially the script make it easy for us to reproduce, and they turn @javafanboy's suspicion into something we can measure and fix. And thank you @javafanboy for raising it back in June; you were right.

Where things stand on main:

• Query on a table: already fixed by your own perf(dynamodb): query only the requested partition #6668, which reads only the requested partition's range of the map instead of the whole table. It's on nightly now and will ship in the next release.
• Query on a secondary index (GSI or LSI): still walks the whole table. That's next: keep a sorted view per index, maintained on each write, and read the partition's range from it the same way.
• Writes in memory mode: every write still copies the whole table, because the in-memory store doesn't write single items. The live map already holds the state, so the plan is to drop that copy and make a write cost the same at 1k items as at 100k.
• Writes in persistent mode: the per-table layout with shards from feat(dynamodb): opt-in per-table item files, split into shards #6303 already covers this on nightly. We'll look at making it the default.

Once those land we'll rerun your script at the same sizes and post the numbers here, so we can see the change against your baseline. If you'd like to pick one of them up, let us know and we'll coordinate so nobody duplicates work.

[Mohith1612]

Thanks @hectorvent! I'd like to take the memory-mode write copy, since the benchmark is already set up to show the before and after. I'll open a PR referencing this thread.