# /// script # requires-python = ">=3.11" # dependencies = ["duckdb>=1.4", "pyiceberg[pyarrow,pyiceberg-core]>=0.10", "boto3", "requests"] # /// """Benchmark audit-log reads: Iceberg on S3 Tables and R2 Data Catalog against DynamoDB. uv run bench.py gen # synthetic events -> data/events.parquet uv run bench.py load {s3tables,r2} --layout sorted # Iceberg table audit.events_ uv run bench.py load dynamodb --rows 1000000 uv run bench.py stats {s3tables,r2} # files per table, e.g. to see compaction finish uv run bench.py query {duckdb-s3tables,duckdb-r2,athena,r2sql,dynamodb} --table events_sorted [--window-days 30] """ import argparse import csv import json import os import statistics import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone from pathlib import Path import duckdb ROOT = Path(__file__).parent DATA = ROOT / "data" / "events.parquet" NAMESPACE = "audit" NOW = "TIMESTAMP '2026-10-01'" # gen spreads events over the year before this # view -> (column the UI filters on, extra predicate that keeps the sample on-topic) VIEWS = { "account": ("owner_account_id", "true"), "product": ("resource_id", "resource_type = 'product'"), "actor": ("actor_id", "true"), } def env(key): if not (value := os.environ.get(key)): raise SystemExit(f"{key} is not set (see .env.example)") return value def gen(a): """Deterministic events; hash(i, salt) stands in for random() so reruns produce the same file.""" DATA.parent.mkdir(exist_ok=True) duckdb.sql(f""" COPY ( WITH r AS ( SELECT i, {", ".join(f"(hash(i, '{s}') % 1000000) / 1e6 AS r{s}" for s in "opuasb")} FROM range({a.rows}) t(i) ), base AS ( SELECT *, -- cubic skew: the busiest account holds ~6% of all events, most accounts very few 'acct' || floor(pow(ro, 3) * {a.accounts})::INT AS owner_account_id, list_element(['object.put', 'object.put', 'object.put', 'object.put', 'object.delete', 'object.copy', 'product.update', 'membership.create', 'membership.delete', 'api_key.create'], floor(ra * 10)::INT + 1) AS action FROM r ) SELECT md5(i::VARCHAR) AS event_id, TIMESTAMP '2025-10-01' + to_microseconds((hash(i, 't') % 31536000000000)::BIGINT) AS ts, -- 365 days of µs action, CASE WHEN action LIKE 'object.%' THEN 'proxy' WHEN rs < 0.7 THEN 'ui' ELSE 'api' END AS source, CASE WHEN ru > 0.9 THEN 'service_account' ELSE 'user' END AS actor_type, -- each account has ~6 regular actors; actors overlap across accounts CASE WHEN ru > 0.9 THEN owner_account_id || '--ci' ELSE 'user' || ((hash(owner_account_id) % {a.users}) + floor(ru * 6)::INT) % {a.users} END AS actor_id, CASE WHEN hash(i, 'imp') % 500 = 0 THEN 'admin' || hash(i, 'adm') % 5 END AS real_actor_id, owner_account_id, CASE WHEN action LIKE 'object.%' OR action LIKE 'product.%' THEN 'product' ELSE 'account' END AS resource_type, CASE WHEN action LIKE 'object.%' OR action LIKE 'product.%' THEN owner_account_id || '/prod' || floor(pow(rp, 2) * 20)::INT ELSE owner_account_id END AS resource_id, md5(i::VARCHAR || 'req') AS request_id, json_object( 'object_key', 'data/' || year(ts) || '/part-' || (hash(i, 'k') % 10000) || '.parquet', 'bytes', (rb * 5e9)::BIGINT, 'etag', md5(i::VARCHAR || 'etag'), 'user_agent', list_element(['aws-cli/2.17.0', 'boto3/1.35.0', 'source-cli/0.4.1', 'Mozilla/5.0'], floor(rb * 4)::INT + 1), 'before', json_object('visibility', 'restricted', 'title', 'Dataset ' || hash(i, 'x') % 1000), 'after', json_object('visibility', 'public', 'title', 'Dataset ' || hash(i, 'y') % 1000) )::VARCHAR AS detail FROM base ORDER BY ts ) TO '{DATA}' (FORMAT parquet) """) print(duckdb.sql(f""" SELECT count(*) AS events, count(DISTINCT owner_account_id) AS accounts, count(DISTINCT resource_id) FILTER (resource_type = 'product') AS products, count(DISTINCT actor_id) AS actors, round(avg(strlen(detail)) + 250) AS approx_bytes_per_event FROM '{DATA}'""")) def iceberg_catalog(target): from pyiceberg.catalog import load_catalog if target == "s3tables": region = env("AWS_REGION") return load_catalog( "s3tables", type="rest", warehouse=env("S3TABLES_ARN"), uri=f"https://s3tables.{region}.amazonaws.com/iceberg", **{"rest.sigv4-enabled": "true", "rest.signing-name": "s3tables", "rest.signing-region": region}, ) return load_catalog("r2", type="rest", warehouse=env("R2_WAREHOUSE"), uri=env("R2_CATALOG_URI"), token=env("R2_TOKEN")) def load_iceberg(a): """`sorted` is what sort-compaction would leave; `stream` is many small time-ordered commits, as Firehose/Pipelines write before compaction.""" import pyarrow as pa from pyiceberg.transforms import MonthTransform catalog = iceberg_catalog(a.target) catalog.create_namespace_if_not_exists(NAMESPACE) ident = (NAMESPACE, a.table or f"events_{a.layout}") if catalog.table_exists(ident): raise SystemExit(f"{'.'.join(ident)} already exists; drop it first") order = "date_trunc('month', ts), owner_account_id, resource_id, ts" if a.layout == "sorted" else "ts" batch_rows = a.batch_rows or (500_000 if a.layout == "sorted" else 10_000) reader = duckdb.sql(f"SELECT * FROM '{DATA}' ORDER BY {order}").fetch_record_batch(batch_rows) table = catalog.create_table(ident, schema=reader.schema, properties={"write.parquet.row-group-limit": str(a.row_group)}) with table.update_spec() as spec: spec.add_field("ts", MonthTransform(), "ts_month") done, t0 = 0, time.time() for batch in reader: table.append(pa.Table.from_batches([batch])) done += batch.num_rows print(f"\r{done:,} rows, {time.time() - t0:.0f}s", end="", flush=True) print(f"\nloaded {'.'.join(ident)} on {a.target}") DDB_KEYS = { # view column -> (index, key attribute, key prefix) "owner_account_id": ("by_owner", "owner_pk", "owner#"), "resource_id": (None, "pk", "resource#product#"), "actor_id": ("by_actor", "actor_pk", "actor#"), } def load_dynamodb(a): import boto3 name = env("DDB_TABLE") client = boto3.client("dynamodb") if name not in client.list_tables()["TableNames"]: client.create_table( TableName=name, BillingMode="PAY_PER_REQUEST", KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}, {"AttributeName": "sk", "KeyType": "RANGE"}], AttributeDefinitions=[{"AttributeName": k, "AttributeType": "S"} for k in ("pk", "sk", "owner_pk", "actor_pk")], GlobalSecondaryIndexes=[ {"IndexName": ix, "Projection": {"ProjectionType": "ALL"}, "KeySchema": [{"AttributeName": key, "KeyType": "HASH"}, {"AttributeName": "sk", "KeyType": "RANGE"}]} for ix, key, _ in DDB_KEYS.values() if ix ], ) client.get_waiter("table_exists").wait(TableName=name) cur = duckdb.sql(f"SELECT * FROM '{DATA}' USING SAMPLE {a.rows} ROWS (reservoir, 42)") cols = cur.columns rows = cur.fetchall() def item(row): e = {k: v for k, v in zip(cols, row) if v is not None} e["ts"] = e["ts"].isoformat() e["sk"] = f"{e['ts']}#{e['event_id']}" e["pk"] = f"resource#{e['resource_type']}#{e['resource_id']}" e["owner_pk"] = f"owner#{e['owner_account_id']}" e["actor_pk"] = f"actor#{e['actor_id']}" return e def write(chunk): table = boto3.session.Session().resource("dynamodb").Table(name) # one session per thread with table.batch_writer() as w: for row in chunk: w.put_item(Item=item(row)) t0, n = time.time(), 16 with ThreadPoolExecutor(n) as ex: list(ex.map(write, [rows[i::n] for i in range(n)])) print(f"loaded {len(rows):,} items into {name} in {time.time() - t0:.0f}s (~${len(rows) * 3 * 0.625e-6:.2f} in writes)") def stats(a): catalog = iceberg_catalog(a.target) for _, name in catalog.list_tables(NAMESPACE): files = catalog.load_table((NAMESPACE, name)).inspect.files().select(["record_count", "file_size_in_bytes"]).to_pylist() size = sum(f["file_size_in_bytes"] for f in files) print(f"{name:20} {len(files):6,} files {size / 1e6:8.1f} MB avg {size / max(len(files), 1) / 1e6:6.1f} MB" f" {sum(f['record_count'] for f in files):,} rows") def load(a): load_dynamodb(a) if a.target == "dynamodb" else load_iceberg(a) def sql(table, col, value, window_days=None): # ponytail: values are generated ids ([a-z0-9/-]), inlined so every engine sees a literal it can prune on since = f" AND ts >= {NOW} - INTERVAL '{window_days}' DAY" if window_days else "" return f"SELECT * FROM {table} WHERE {col} = '{value}'{since} ORDER BY ts DESC LIMIT 50" def duckdb_runner(target, table): con = duckdb.connect() con.execute("INSTALL iceberg; LOAD iceberg; INSTALL httpfs; LOAD httpfs; INSTALL aws; LOAD aws") # a serverless function can't count on the previous request's reads, so every query fetches its data from the store con.execute("SET enable_external_file_cache = false") if target == "s3tables": con.execute(f"CREATE SECRET (TYPE s3, PROVIDER credential_chain, REGION '{env('AWS_REGION')}')") con.execute(f"ATTACH '{env('S3TABLES_ARN')}' AS cat (TYPE iceberg, ENDPOINT_TYPE s3_tables)") else: con.execute(f"CREATE SECRET r2 (TYPE iceberg, TOKEN '{env('R2_TOKEN')}')") con.execute(f"ATTACH '{env('R2_WAREHOUSE')}' AS cat (TYPE iceberg, ENDPOINT '{env('R2_CATALOG_URI')}', SECRET r2)") return lambda col, v, w: (len(con.execute(sql(f"cat.{NAMESPACE}.{table}", col, v, w)).fetchall()), None) def athena_runner(table): import boto3 athena = boto3.client("athena") bucket = env("S3TABLES_ARN").rsplit("/", 1)[1] ctx = {"Catalog": f"s3tablescatalog/{bucket}", "Database": NAMESPACE} def run(col, v, w): qid = athena.start_query_execution( QueryString=sql(f'"{table}"', col, v, w), QueryExecutionContext=ctx, ResultConfiguration={"OutputLocation": env("ATHENA_OUTPUT")}, )["QueryExecutionId"] while (q := athena.get_query_execution(QueryExecutionId=qid)["QueryExecution"])["Status"]["State"] in ("QUEUED", "RUNNING"): time.sleep(0.05) if q["Status"]["State"] != "SUCCEEDED": raise RuntimeError(q["Status"]) rows = athena.get_query_results(QueryExecutionId=qid)["ResultSet"]["Rows"] return len(rows) - 1, q["Statistics"].get("DataScannedInBytes") return run def r2sql_runner(table): import requests url = f"https://api.sql.cloudflarestorage.com/api/v1/accounts/{env('CF_ACCOUNT_ID')}/basin-sql/query/{env('R2_BUCKET')}" session = requests.Session() session.headers["Authorization"] = f"Bearer {env('R2_TOKEN')}" shown = False def run(col, v, w): nonlocal shown resp = session.post(url, json={"query": sql(f"{NAMESPACE}.{table}", col, v, w)}) if not resp.ok: # e.g. a 400 when R2 SQL estimates the scan is over its budget raise RuntimeError(f"{resp.status_code} {resp.text[:200]}") body = resp.json() if not shown: # the response shape isn't documented; show one so the row count below can be checked print("first R2 SQL response:", json.dumps(body)[:400]) shown = True result = body.get("result") or {} rows = result.get("rows") if isinstance(result, dict) else result return (len(rows) if isinstance(rows, list) else -1), None return run def dynamodb_runner(_table): import boto3 from boto3.dynamodb.conditions import Key table = boto3.resource("dynamodb").Table(env("DDB_TABLE")) def run(col, v, w): index, key, prefix = DDB_KEYS[col] cond = Key(key).eq(prefix + v) if w: # sk starts with the ISO timestamp, so the window is a key range cond &= Key("sk").gte((datetime(2026, 10, 1) - timedelta(days=w)).isoformat()) kwargs = {"IndexName": index} if index else {} resp = table.query(KeyConditionExpression=cond, ScanIndexForward=False, Limit=50, **kwargs) return len(resp["Items"]), None return run RUNNERS = { "duckdb-s3tables": lambda t: duckdb_runner("s3tables", t), "duckdb-r2": lambda t: duckdb_runner("r2", t), "athena": athena_runner, "r2sql": r2sql_runner, "dynamodb": dynamodb_runner, } def query(a): """Targets are sampled per event, so busy accounts come up as often as they would in real traffic.""" run = RUNNERS[a.engine](a.table) out = ROOT / "results.csv" new = not out.exists() at = datetime.now(timezone.utc).isoformat(timespec="seconds") with out.open("a", newline="") as f: w = csv.writer(f) if new: w.writerow(["at", "engine", "table", "window_days", "view", "run", "target", "ms", "rows", "bytes_scanned", "error"]) since = f" AND ts >= {NOW} - INTERVAL '{a.window_days}' DAY" if a.window_days else "" for view, (col, where) in VIEWS.items(): targets = [r[0] for r in duckdb.sql( f"SELECT {col} FROM (SELECT * FROM '{DATA}' WHERE {where}{since}) USING SAMPLE {a.runs} ROWS (reservoir, 7)").fetchall()] ms, errors = [], 0 for i, target in enumerate(targets): t0 = time.perf_counter() try: rows, scanned, error = *run(col, target, a.window_days), None except Exception as e: # recorded, not fatal: a rejected query is a result too rows, scanned, error = None, None, str(e)[:300] errors += 1 ms.append((time.perf_counter() - t0) * 1000) w.writerow([at, a.engine, a.table, a.window_days, view, i, target, round(ms[-1], 1), rows, scanned, error]) warm = ms[1:] q = statistics.quantiles(warm, n=100, method="inclusive") # never beyond the observed max print(f"{a.engine:16} {a.table:16} {view:8} first {ms[0]:6.0f} p50 {q[49]:6.0f} p95 {q[94]:6.0f}" f" max {max(warm):6.0f} ms errors {errors}/{len(ms)}") def main(): p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) sub = p.add_subparsers(required=True) g = sub.add_parser("gen", help="write data/events.parquet") g.add_argument("--rows", type=int, default=5_000_000) g.add_argument("--accounts", type=int, default=5_000) g.add_argument("--users", type=int, default=20_000) g.set_defaults(fn=gen) l = sub.add_parser("load", help="load events into a store") l.add_argument("target", choices=["s3tables", "r2", "dynamodb"]) l.add_argument("--layout", choices=["sorted", "stream"], default="sorted") l.add_argument("--table", help="table name (default events_)") l.add_argument("--batch-rows", type=int, help="rows per Iceberg commit (default 500k sorted, 10k stream)") l.add_argument("--row-group", type=int, default=20_000, help="Parquet row-group size; smaller prunes finer") l.add_argument("--rows", type=int, default=1_000_000, help="DynamoDB only: how many events to sample") l.set_defaults(fn=load) s = sub.add_parser("stats", help="files, size and rows per Iceberg table") s.add_argument("target", choices=["s3tables", "r2"]) s.set_defaults(fn=stats) q = sub.add_parser("query", help="time the three views; appends to results.csv") q.add_argument("engine", choices=RUNNERS) q.add_argument("--table", default="events_sorted") q.add_argument("--runs", type=int, default=200, help="queries per view") q.add_argument("--window-days", type=int, help="only events in the last N days, as a default Activity tab would show") q.set_defaults(fn=query) a = p.parse_args() a.fn(a) if __name__ == "__main__": main()