Benchmarking Core Executemany Bulk Insert Performance

To benchmark SQLAlchemy Core executemany bulk inserts reliably, wrap each strategy in a warmed-up time.perf_counter() loop over a fixed row set, then cross-check the server-side cost in pg_stat_statements — never trust a single cold run. This page is the measurement companion to the parent guide on high-performance bulk inserts and updates, which covers the upsert, RETURNING, and synchronize_session strategies you will be benchmarking here.

Isolating variables one at a time — batch size, row width, insertmanyvalues_page_size, index count — and confirming the actual SQL shape with EXPLAIN (ANALYZE, BUFFERS) is what turns anecdotal "asyncpg is fast" claims into reproducible rows-per-second numbers you can defend in a capacity review. Because raw insert speed is bounded by driver encoding and pool contention, benchmark only after you have configured your async engine and connection pool the same way production runs it.

Quick Answer

# Naive benchmark — one cold run, times everything including compile + cache miss
import time
t0 = time.perf_counter()
await conn.execute(insert(orders_table), rows)  # first call also pays statement compilation
print(time.perf_counter() - t0)                 # misleading: 2-5x slower than steady state

# Reliable benchmark — pre-build rows, warm up, take the median of N runs
import asyncio
import time
from statistics import median
from sqlalchemy import insert, text, Table, MetaData, Column, Integer, String, Numeric
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine

metadata = MetaData()
orders_table = Table(
    "orders", metadata,
    Column("id", Integer, primary_key=True),
    Column("customer_id", Integer, nullable=False),
    Column("amount", Numeric(12, 4), nullable=False),
    Column("status", String(32), nullable=False),
)

async def measure_insert(engine: AsyncEngine, rows: list[dict], label: str, runs: int = 5) -> float:
    stmt = insert(orders_table)
    samples: list[float] = []
    for i in range(runs + 1):                       # +1 warm-up run, discarded below
        async with engine.begin() as conn:
            await conn.execute(text("TRUNCATE orders RESTART IDENTITY"))
            t0 = time.perf_counter()
            await conn.execute(stmt, rows)          # dict list pre-built outside the timed block
            samples.append(time.perf_counter() - t0)
    steady = median(samples[1:])                    # drop the cold first sample
    rps = len(rows) / steady
    print(f"{label}: {rps:,.0f} rows/sec ({steady:.3f}s median for {len(rows):,} rows)")
    return rps

# Run: asyncio.run(measure_insert(engine, rows, "Core executemany"))

The single-run form on top is the mistake almost every ad-hoc benchmark makes: the first execute() also pays SQLAlchemy statement compilation and a cold buffer cache, so it reports throughput two-to-five times worse than steady state. The median-of-runs form below is the pattern the rest of this page builds on.

Execution Context & Async Workflow Integration

What the Benchmark Is Actually Measuring

A naive time.perf_counter() wrapper around a single conn.execute() call measures end-to-end wall time including:

• Python list construction and dict iteration
• SQLAlchemy statement compilation (cached after first call)
• asyncpg binary parameter encoding
• TCP round-trip(s) to the database server
• PostgreSQL WAL write and index maintenance
• Python-side result fetch (if RETURNING is used)

To isolate the SQLAlchemy + driver overhead from PostgreSQL server overhead, run the benchmark with the database on localhost (Unix socket preferred), then repeat on a production-representative network. The delta quantifies network latency's contribution and tells you whether further Python-side optimization will matter. The choice of driver dominates the encoding cost measured here, so a fair comparison depends on choosing between asyncpg and psycopg deliberately rather than benchmarking whichever one happened to be installed.

Benchmark Setup: Consistent Baseline

Before comparing strategies, establish a consistent baseline environment:

import asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncEngine

async def prepare_benchmark(engine: AsyncEngine, target_rows: int = 100_000) -> list[dict]:
    """Reset table and generate synthetic rows for benchmark."""
    async with engine.begin() as conn:
        await conn.execute(text("TRUNCATE TABLE orders RESTART IDENTITY"))

    return [
        {
            "customer_id": i % 10_000,
            "amount": f"{(i % 500) + 0.99:.2f}",
            "status": "pending" if i % 3 else "completed",
        }
        for i in range(target_rows)
    ]

Always TRUNCATE between runs to avoid index bloat from prior inserts skewing subsequent measurements. Restart the sequence to prevent PK gaps that inflate index page splits.

Strategy Comparison with time.perf_counter

import asyncio
import time
from itertools import islice
from sqlalchemy import insert
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker

class Base(DeclarativeBase):
    pass

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int]
    amount: Mapped[str]   # simplified for benchmark
    status: Mapped[str]

async def benchmark_all(engine: AsyncEngine, rows: list[dict]) -> None:
    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

    # --- Strategy 1: ORM add_all ---
    instances = [Order(**r) for r in rows]
    async with SessionLocal() as session:
        t0 = time.perf_counter()
        session.add_all(instances)
        await session.commit()
        elapsed = time.perf_counter() - t0
    print(f"ORM add_all:          {len(rows)/elapsed:>10,.0f} rows/s")

    async with engine.begin() as conn:
        await conn.execute(__import__("sqlalchemy").text("TRUNCATE orders RESTART IDENTITY"))

    # --- Strategy 2: ORM bulk execute (session.execute + insert) ---
    async with SessionLocal() as session:
        t0 = time.perf_counter()
        await session.execute(insert(Order), rows)
        await session.commit()
        elapsed = time.perf_counter() - t0
    print(f"ORM bulk execute:     {len(rows)/elapsed:>10,.0f} rows/s")

    async with engine.begin() as conn:
        await conn.execute(__import__("sqlalchemy").text("TRUNCATE orders RESTART IDENTITY"))

    # --- Strategy 3: Core executemany ---
    stmt = insert(orders_table)
    t0 = time.perf_counter()
    async with engine.begin() as conn:
        await conn.execute(stmt, rows)
    elapsed = time.perf_counter() - t0
    print(f"Core executemany:     {len(rows)/elapsed:>10,.0f} rows/s")

    async with engine.begin() as conn:
        await conn.execute(__import__("sqlalchemy").text("TRUNCATE orders RESTART IDENTITY"))

    # --- Strategy 4: insertmanyvalues (RETURNING) ---
    stmt_ret = insert(orders_table).returning(orders_table.c.id)
    t0 = time.perf_counter()
    async with engine.begin() as conn:
        result = await conn.execute(stmt_ret, rows)
        _ = result.fetchall()
    elapsed = time.perf_counter() - t0
    print(f"insertmanyvalues:     {len(rows)/elapsed:>10,.0f} rows/s")

Representative Throughput Numbers

These numbers were measured on a c5.xlarge EC2 instance connecting to db.r6g.xlarge RDS PostgreSQL 15 via private subnet, with 100,000 rows of the shape above (3 non-PK columns, no foreign keys, no extra indexes). The absolute spread between the ORM unit-of-work and the raw COPY path is roughly 40x, which is why the Core vs ORM overhead decision matters far more for ingestion than for OLTP:

Add indexes and foreign keys and all numbers drop 20–60% depending on index selectivity. Profile your specific schema rather than relying on synthetic benchmarks.

Resolving Warnings, Errors & Common Mistakes

Advanced Benchmarking Optimization

Using EXPLAIN (ANALYZE, BUFFERS) to Confirm Strategy

SQLAlchemy does not expose what SQL it actually sent for insertmanyvalues in production logs. Use echo=True on the engine temporarily, or inspect pg_stat_statements to verify the rewritten statement shape:

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncEngine

async def inspect_insert_statements(engine: AsyncEngine) -> None:
    """Print the top insert statements from pg_stat_statements by total_time."""
    async with engine.connect() as conn:
        result = await conn.execute(text("""
            SELECT
                left(query, 120) AS query_snippet,
                calls,
                round((total_exec_time / calls)::numeric, 2) AS avg_ms,
                rows / calls AS avg_rows
            FROM pg_stat_statements
            WHERE query ILIKE '%INSERT%orders%'
            ORDER BY total_exec_time DESC
            LIMIT 10
        """))
        for row in result:
            print(f"avg {row.avg_ms}ms | {row.avg_rows} rows/call | {row.query_snippet}")

insertmanyvalues rewrites the single-row INSERT INTO orders (customer_id, amount, status) VALUES ($1, $2, $3) into multi-value form. You should see one statement with high avg_rows (up to insertmanyvalues_page_size) rather than many single-row statements.

Varying Batch Sizes Systematically

Throughput is not linear with batch size. Run a sweep across chunk sizes to find the elbow:

import asyncio
import time
from sqlalchemy import insert
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from sqlalchemy import text

async def batch_size_sweep(engine: AsyncEngine, all_rows: list[dict]) -> None:
    stmt = insert(orders_table)
    for chunk_size in [100, 500, 1_000, 2_000, 5_000, 10_000, 25_000]:
        async with engine.begin() as conn:
            await conn.execute(text("TRUNCATE orders RESTART IDENTITY"))

        t0 = time.perf_counter()
        async with engine.begin() as conn:
            for i in range(0, len(all_rows), chunk_size):
                await conn.execute(stmt, all_rows[i:i + chunk_size])
        elapsed = time.perf_counter() - t0
        print(f"chunk={chunk_size:>6}: {len(all_rows)/elapsed:>10,.0f} rows/s")

Typical results: throughput rises steeply from 100 to ~2,000 rows per chunk, then plateaus or slightly decreases above 10,000 due to memory allocation and WAL pressure. The optimal chunk size for your workload sits at the plateau's left edge. When your source data is larger than memory, feed the sweep from a generator rather than a materialized list — the memory-safe chunking pattern is covered in batch inserting millions of rows with SQLAlchemy Core.

Advanced Profiling Techniques

Correlating Python Timing with pg_stat_statements

Python-side time.perf_counter() measures wall time including event loop scheduling latency, GIL contention, and network jitter. For a clean server-side view, cross-reference with pg_stat_statements:

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncEngine

async def server_side_timing(engine: AsyncEngine, label: str) -> None:
    """Print server-side execution stats for recent INSERT statements."""
    async with engine.connect() as conn:
        result = await conn.execute(text("""
            SELECT
                left(query, 100)      AS query_preview,
                calls,
                round(mean_exec_time::numeric, 2) AS mean_ms,
                rows / calls          AS rows_per_call
            FROM pg_stat_statements
            WHERE query ILIKE '%INSERT%orders%'
            ORDER BY total_exec_time DESC
            LIMIT 5
        """))
        print(f"\n=== {label} ===")
        for row in result:
            print(f"  {row.mean_ms}ms avg | {row.rows_per_call} rows/call | {row.query_preview}")

Run SELECT pg_stat_statements_reset() before each benchmark strategy to isolate its statistics. If rows_per_call is 1 for an insert you expect to be batched, insertmanyvalues is not activating — check that .returning() is present and the dialect supports it.

Profiling Index Maintenance Cost

Index maintenance during bulk inserts is often the dominant cost for tables with non-trivial indexes. Isolate this by benchmarking with and without indexes:

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncEngine
import time

async def benchmark_with_without_indexes(engine: AsyncEngine, rows: list[dict]) -> None:
    stmt = __import__("sqlalchemy").insert(orders_table)

    # Without indexes
    async with engine.begin() as conn:
        await conn.execute(text("DROP INDEX IF EXISTS idx_orders_customer"))
        await conn.execute(text("TRUNCATE orders RESTART IDENTITY"))
    t0 = time.perf_counter()
    async with engine.begin() as conn:
        await conn.execute(stmt, rows)
    no_idx_time = time.perf_counter() - t0

    # With index
    async with engine.begin() as conn:
        await conn.execute(text("CREATE INDEX idx_orders_customer ON orders(customer_id)"))
        await conn.execute(text("TRUNCATE orders RESTART IDENTITY"))
    t0 = time.perf_counter()
    async with engine.begin() as conn:
        await conn.execute(stmt, rows)
    with_idx_time = time.perf_counter() - t0

    overhead_pct = (with_idx_time - no_idx_time) / no_idx_time * 100
    print(f"Index overhead: {overhead_pct:.1f}% slower ({no_idx_time:.2f}s → {with_idx_time:.2f}s)")

A single B-tree index on a narrow column typically adds 20–40% overhead. For tables with 5+ indexes, the index maintenance cost can exceed the raw insert cost. In those cases, consider dropping indexes before bulk load and rebuilding with CREATE INDEX CONCURRENTLY after commit — a standard ETL pattern.

Frequently Asked Questions

Related

• High-Performance Bulk Inserts and Updates — parent guide to all bulk strategies including ON CONFLICT upsert, synchronize_session, and asyncpg COPY.
• Batch Inserting Millions of Rows with SQLAlchemy Core — memory-safe generator patterns and per-chunk transaction management for production-scale inserts.
• Streaming Large Result Sets with yield_per — the read-side counterpart: benchmarking and tuning throughput when pulling millions of rows back out.
• Configuring Async Engines and Connection Pools — pool sizing and driver configuration that directly affects bulk insert benchmark results.