Assess performance of ORM, raw SQL with ID column, and raw SQL without ID column.

ORM-performance-overhead-assessment-commands-and-output.md

In-Memory Sqlite

1. ORM: python3.12 orm-tester.py sqlite 100000
   • 100000 additions took 27251.844822ms
2. Raw SQL: python3.12 sqlite-tester.py 100000
   • With ID column: 100000 additions took 6531.315638ms
   • Without ID column: 100000 additions took 6026.203085ms

Postgres

1. docker run -e POSTGRES_PASSWORD=random -e POSTGRES_USER=postgres -e POSTGRES_DB=test-db -p 5432:5432 postgres:18-alpine
2. ORM: python3.12 orm-tester.py postgres 100000
   • 100000 additions took 204872.66124ms
3. Raw SQL: python3.12 postgres-tester.py 100000
   • With ID column: 100000 additions took 161722.944119ms
   • Without ID column: 100000 additions took 144713.248249ms

orm-tester.py

import datetime
import sys
import time
from sqlalchemy import Engine, create_engine, func, text, String
from sqlalchemy.orm import Session, DeclarativeBase, MappedAsDataclass, Mapped, mapped_column

_TEST_TABLE = "test_table"

class Base(DeclarativeBase, MappedAsDataclass):
    pass

class TestData(Base):
    __tablename__ = _TEST_TABLE

    host: Mapped[str] = mapped_column(String[10])
    status: Mapped[int]
    id: Mapped[int] = mapped_column(default=None, primary_key=True)
    time: Mapped[datetime.datetime] = mapped_column(server_default=func.now(),
                                                    default_factory=datetime.datetime.now)

def _delete_table(engine: Engine) -> None:
    with Session(engine) as session:
        session.execute(text(f"DROP TABLE IF EXISTS {_TEST_TABLE}"))
        session.commit()

def _create_table(engine: Engine) -> None:
    Base.metadata.create_all(engine)

def _add_data_to_table(engine: Engine, count: int) -> None:
    with Session(engine) as session:
        for i in range(count):
            tmp = TestData(host=str(i), status=i)
            session.add(tmp)
            session.commit()
            if not (i % 10000) and i:
                print(f"Completed {i} additions")

if __name__ == "__main__":
    match sys.argv[1]:
        case "sqlite":
            engine = create_engine("sqlite:///:memory:")
        case "postgres":
            engine = create_engine("postgresql+psycopg://postgres:random@localhost/test-db")
        case _ :
            raise ValueError("""Please provide "sqlite" or "postgres" as the first argument""")

    _delete_table(engine)
    _create_table(engine)

    count = int(sys.argv[2])
    start = time.monotonic_ns()
    _add_data_to_table(engine, count)
    stop = time.monotonic_ns()
    print(f"{count} additions took {(stop - start) / 1_000_000}ms")

postgres-tester.py

import datetime
import psycopg
import sys
import time
from pypika import Query, Table


_TEST_TABLE_WITH_ID = "test_table_with_id"
_TEST_TABLE_WITHOUT_ID = "test_table_without_id"

def _delete_table(conn: psycopg.Connection) -> None:
    conn.execute(f"""
        DROP TABLE IF EXISTS {_TEST_TABLE_WITH_ID};
        DROP TABLE IF EXISTS {_TEST_TABLE_WITHOUT_ID};
    """)
    conn.commit()

def _create_tables(conn: psycopg.Connection) -> None:
    conn.execute(f"""
    CREATE TABLE test_table_with_id (
    	host VARCHAR NOT NULL,
        status INTEGER NOT NULL,
        id SERIAL NOT NULL,
        time TIMESTAMP NOT NULL,
        PRIMARY KEY (id)
    );
    CREATE TABLE test_table_without_id (
    	host VARCHAR NOT NULL,
        status INTEGER NOT NULL,
        time TIMESTAMP NOT NULL
    );
    """)
    conn.commit()

def _add_data_to_table(conn: psycopg.Connection, count: int, with_id: bool) -> None:
    if with_id:
        table = Table(_TEST_TABLE_WITH_ID)
        q = lambda i: Query.into(table).insert(str(i), i, i, datetime.datetime.now())
    else:
        table = Table(_TEST_TABLE_WITHOUT_ID)
        q = lambda i: Query.into(table).insert(str(i), i, datetime.datetime.now())

    for i in range(count):
        conn.execute(q(i).get_sql())
        conn.commit()
        if not (i % 10000) and i:
            print(f"Completed {i} additions")

if __name__ == "__main__":
    with psycopg.connect("postgresql://postgres:random@localhost/test-db") as conn:
        _delete_table(conn)
        _create_tables(conn)

        count = int(sys.argv[1])
        start = time.monotonic_ns()
        _add_data_to_table(conn, count, True)
        stop = time.monotonic_ns()
        print(f"With ID column: {count} additions took {(stop - start) / 1_000_000}ms")

        start = time.monotonic_ns()
        _add_data_to_table(conn, count, False)
        stop = time.monotonic_ns()
        print(f"Without ID column: {count} additions took {(stop - start) / 1_000_000}ms")

sqlite-tester.py

import datetime
import sqlite3
import sys
import time
from pypika import Query, Table


_TEST_TABLE_WITH_ID = "test_table_with_id"
_TEST_TABLE_WITHOUT_ID = "test_table_without_id"

def _delete_table(conn: sqlite3.Connection) -> None:
    conn.executescript(f"""
        DROP TABLE IF EXISTS {_TEST_TABLE_WITH_ID};
        DROP TABLE IF EXISTS {_TEST_TABLE_WITHOUT_ID};
    """)
    conn.commit()

def _create_tables(conn: sqlite3.Connection) -> None:
    conn.executescript(f"""
    CREATE TABLE test_table_with_id (
        host VARCHAR NOT NULL,
        status INTEGER NOT NULL,
        id INTEGER NOT NULL,
        time DATETIME NOT NULL,
        PRIMARY KEY (id)
    );
    CREATE TABLE test_table_without_id (
        host VARCHAR NOT NULL,
        status INTEGER NOT NULL,
        time DATETIME NOT NULL
    );
    """)
    conn.commit()

def _add_data_to_table(conn: sqlite3.Connection, count: int, with_id: bool) -> None:
    if with_id:
        table = Table(_TEST_TABLE_WITH_ID)
        q = lambda i: Query.into(table).insert(str(i), i, i, datetime.datetime.now())
    else:
        table = Table(_TEST_TABLE_WITHOUT_ID)
        q = lambda i: Query.into(table).insert(str(i), i, datetime.datetime.now())

    for i in range(count):
        conn.execute(q(i).get_sql())
        conn.commit()
        if not (i % 10000) and i:
            print(f"Completed {i} additions")

if __name__ == "__main__":
    with sqlite3.connect(":memory:") as conn:
        _delete_table(conn)
        _create_tables(conn)

        count = int(sys.argv[1])
        start = time.monotonic_ns()
        _add_data_to_table(conn, count, True)
        stop = time.monotonic_ns()
        print(f"With ID column: {count} additions took {(stop - start) / 1_000_000}ms")

        start = time.monotonic_ns()
        _add_data_to_table(conn, count, False)
        stop = time.monotonic_ns()
        print(f"Without ID column: {count} additions took {(stop - start) / 1_000_000}ms")