Benchmarking Core Executemany Bulk Insert Performance

Benchmark SQLAlchemy bulk inserts by wrapping each strategy in time.perf_counter() loops, isolating variables (batch size, row width, insertmanyvalues_page_size, executemany_mode), and confirming actual SQL shapes with EXPLAIN (ANALYZE, BUFFERS) — this gives reproducible throughput numbers rather than anecdotal estimates. All methodologies and production numbers below map back to the strategies covered in High-Performance Bulk Inserts and Updates.

Quick Answer

# Minimal benchmark harness — run each strategy, compare rows/second
import asyncio
import time
from sqlalchemy import insert, 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) -> float:
    stmt = insert(orders_table)
    t0 = time.perf_counter()
    async with engine.begin() as conn:
        await conn.execute(stmt, rows)
    elapsed = time.perf_counter() - t0
    rps = len(rows) / elapsed
    print(f"{label}: {rps:,.0f} rows/sec ({elapsed:.3f}s for {len(rows):,} rows)")
    return rps

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

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.

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):

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.

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

How do I measure only the database time, excluding Python serialization? Record the time immediately before and after await conn.execute(stmt, chunk), with the dict list pre-built outside the timed block. For server-side measurement, use pg_stat_statements.mean_exec_time per call — this excludes all client-side overhead including network RTT and Python result parsing.

Should I use executemany_mode in SQLAlchemy 2.0? No. executemany_mode was an asyncpg-dialect-level parameter in SQLAlchemy 1.4. In 2.0 it is replaced by insertmanyvalues, which is dialect-agnostic and automatically activated when RETURNING is present. If you see executemany_mode in existing code, remove it during your 1.4 → 2.0 migration.

What is a realistic baseline to expect on RDS PostgreSQL? On db.r6g.xlarge RDS over a private VPC connection, Core executemany with asyncpg typically delivers 70,000–120,000 rows/sec for 3–5 column rows without extra indexes. Add two additional indexes and expect a 30–50% reduction. asyncpg COPY on the same hardware delivers 350,000–500,000 rows/sec. These ranges widen significantly with row width, index selectivity, and network RTT.

How do I ensure my benchmark reflects production conditions? Use the same instance class, same Postgres version, same indexes, same autovacuum settings, and same connection pool configuration as production. Run the benchmark during a simulated concurrent read load — bulk inserts under concurrent OLTP traffic behave differently than in an idle database due to lock contention and buffer cache pressure.

Why does my benchmark show higher throughput on the second run than the first? PostgreSQL's buffer cache is cold on the first run — the table and index pages must be loaded from disk. Subsequent runs find those pages already in shared buffers and operate entirely in memory. Always warm up with at least one full pass before recording results, and consider running benchmarks with pg_prewarm to pre-load table pages for a controlled comparison.

Related

• High-Performance Bulk Inserts and Updates — complete 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.
• Configuring Async Engines and Connection Pools — pool sizing and driver configuration that directly affects bulk insert benchmark results.