Indexing EPCIS Events for Sub-Second Verification

A trading partner’s verification request names one GTIN and one serial number and expects a disposition back before its own response-time SLA expires — regardless of whether your repository holds ten million rows or ten billion. That constraint is the exact problem this page solves: designing a schema, index, and partitioning strategy so a lookup for “what is the current status of this serialized unit” runs in milliseconds no matter how large the table grows. It sits inside EPCIS Repository Storage & Retention — the guides covering how commissioned, shipped, and decommissioned events are stored and kept queryable for their full regulatory lifetime — within the broader Serialization Data Ingestion & EPCIS Event Sync area. Get the index design wrong and every verification query degenerates into a sequential scan over years of ObjectEvent history; get it right and the repository answers the same query in constant time as it grows from millions to billions of rows.

Prerequisites #

• Python 3.10+ — the async query examples use X | None union types.
• SQLAlchemy 2.0 — the Mapped/mapped_column declarative style and the async ORM session (AsyncSession, AsyncEngine) used throughout.
• asyncpg (via sqlalchemy[asyncio] with the postgresql+asyncpg dialect) — the async driver behind the query examples.
• PostgreSQL 14+ — native declarative RANGE partitioning, CREATE INDEX ... INCLUDE covering indexes, and index-only scans. The same design translates to other engines that support range partitioning and covering indexes, with adjusted DDL syntax.
• DSCSA data prerequisites — an EPCIS repository already receiving validated ObjectEvent and AggregationEvent records with a GTIN (01), serial (21), event_time, biz_step, and disposition on every row; this page assumes ingestion and schema validation already happened upstream.

Before touching indexes, be precise about the query this design optimizes for: given one GTIN and one serial number, return the most recent event — the one whose disposition and biz_step represent the unit’s current state. That is the query a Verification Router Service issues on every trading-partner request, and it is different from an analytical query that aggregates across many serials. Optimizing for the wrong access pattern is the single most common reason a repository that performs well in testing collapses under production verification load.

Step-by-Step Solution #

Step 1 — Model the event table with SQLAlchemy 2.0 #

Declare the repository table with the fields a verification lookup and an audit reconstruction both need. Because PostgreSQL requires the partition key to be part of any unique constraint, event_time joins gtin and serial_number in the primary key rather than being a bolt-on column.

from __future__ import annotations

from datetime import date, datetime

from sqlalchemy import Date, DateTime, Index, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class EpcisEvent(Base):
    """One EPCIS event row (ObjectEvent or AggregationEvent) in the repository."""

    __tablename__ = "epcis_events"
    __table_args__ = (
        Index(
            "ix_epcis_events_gtin_serial_time",
            "gtin",
            "serial_number",
            "event_time",
            postgresql_ops={"event_time": "DESC"},
            postgresql_include=["disposition", "biz_step", "event_type"],
        ),
        {"postgresql_partition_by": "RANGE (event_time)"},
    )

    gtin: Mapped[str] = mapped_column(String(14), primary_key=True)
    serial_number: Mapped[str] = mapped_column(String(20), primary_key=True)
    event_time: Mapped[datetime] = mapped_column(DateTime(timezone=True), primary_key=True)
    event_type: Mapped[str] = mapped_column(String(32))
    biz_step: Mapped[str] = mapped_column(String(128))
    disposition: Mapped[str] = mapped_column(String(128))
    read_point: Mapped[str] = mapped_column(String(128))
    lot_number: Mapped[str | None] = mapped_column(String(20), nullable=True)
    expiration_date: Mapped[date | None] = mapped_column(Date, nullable=True)

DSCSA/GS1 note: GTIN (01) and serial (21) together form the SGTIN that DSCSA verification keys on, and keeping both as native primary-key columns — not buried inside a single EPC URI string — is what lets the planner use a native composite index instead of parsing text on every row.

Step 2 — Add the composite covering index #

The index in Step 1’s __table_args__ puts gtin and serial_number first, event_time DESC last so the most recent row for a unit sorts to the front of its index range, and lists disposition, biz_step, and event_type as INCLUDE columns. That last part matters more than it looks: without it, PostgreSQL still has to fetch the heap page for every candidate row to read the disposition. With the included columns, an index-only scan answers the query from the index alone.

CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_epcis_events_gtin_serial_time
    ON epcis_events (gtin, serial_number, event_time DESC)
    INCLUDE (disposition, biz_step, event_type);

DSCSA/GS1 note: column order is not cosmetic — (gtin, serial, event_time) supports point lookups for one serial and range scans across every serial under a GTIN, but it will not accelerate a query that filters on serial_number alone without a gtin, because the index’s leftmost prefix rule skips straight past that column.

Step 3 — Partition the repository by event_time #

A single unpartitioned table with billions of rows still means the index itself, and the working set the planner touches, keeps growing forever. Monthly range partitioning bounds both: each partition carries its own copy of the composite index, so a query that also constrains event_time prunes to one partition before the index is even touched.

CREATE TABLE epcis_events (
    gtin            VARCHAR(14) NOT NULL,
    serial_number   VARCHAR(20) NOT NULL,
    event_time      TIMESTAMPTZ NOT NULL,
    event_type      VARCHAR(32) NOT NULL,
    biz_step        VARCHAR(128) NOT NULL,
    disposition     VARCHAR(128) NOT NULL,
    read_point      VARCHAR(128) NOT NULL,
    lot_number      VARCHAR(20),
    expiration_date DATE,
    PRIMARY KEY (gtin, serial_number, event_time)
) PARTITION BY RANGE (event_time);

CREATE TABLE epcis_events_2026_07
    PARTITION OF epcis_events
    FOR VALUES FROM ('2026-07-01') TO ('2026-08-01');

CREATE INDEX CONCURRENTLY ix_epcis_events_2026_07_gtin_serial_time
    ON epcis_events_2026_07 (gtin, serial_number, event_time DESC)
    INCLUDE (disposition, biz_step, event_type);

DSCSA/GS1 note: six years of monthly partitions is 72 partitions per year of retained history, a scale that stays manageable under the six-year window that Enforcing Six-Year Retention with Python automates — that guide covers rolling new partitions on, detaching expired ones, and archiving them rather than deleting rows one at a time.

Step 4 — Write the async “current status” query #

A trading-partner verification call cares about one thing: the newest row for one GTIN-serial pair. Ordering by event_time DESC and taking the first row lets the composite index answer the whole query without a sort step, because the index already stores rows in that order.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession


async def get_current_status(
    session: AsyncSession, gtin: str, serial_number: str
) -> dict[str, object] | None:
    """Return the latest disposition for one serialized unit, or None if unseen."""
    stmt = (
        select(
            EpcisEvent.event_time,
            EpcisEvent.disposition,
            EpcisEvent.biz_step,
            EpcisEvent.event_type,
        )
        .where(
            EpcisEvent.gtin == gtin,
            EpcisEvent.serial_number == serial_number,
        )
        .order_by(EpcisEvent.event_time.desc())
        .limit(1)
    )
    result = await session.execute(stmt)
    row = result.first()
    if row is None:
        return None
    return {
        "gtin": gtin,
        "serial_number": serial_number,
        "event_time": row.event_time,
        "disposition": row.disposition,
        "biz_step": row.biz_step,
        "event_type": row.event_type,
    }

DSCSA/GS1 note: this is the query shape a Verification Router Service implementation calls on every partner request, so its p99 latency directly determines whether the router meets its federally-driven response-time obligation; a query plan that falls back to a sequential scan under load is a compliance risk, not just a performance one.

Step 5 — Confirm the plan uses the index, not a scan #

Trust the query plan, not the query text. Run EXPLAIN (ANALYZE, BUFFERS) against production-representative data and confirm the plan shows Index Only Scan on the composite index rather than Seq Scan or a Bitmap Heap Scan that still hits the base table.

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession


async def explain_current_status(
    session: AsyncSession, gtin: str, serial_number: str
) -> list[str]:
    """Return the EXPLAIN ANALYZE plan lines for the current-status query."""
    plan_sql = text(
        """
        EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
        SELECT event_time, disposition, biz_step, event_type
        FROM epcis_events
        WHERE gtin = :gtin AND serial_number = :serial_number
        ORDER BY event_time DESC
        LIMIT 1
        """
    )
    result = await session.execute(plan_sql, {"gtin": gtin, "serial_number": serial_number})
    return [row[0] for row in result.fetchall()]

DSCSA/GS1 note: keep this check in a pre-production runbook — a schema migration, a dropped INCLUDE column, or a partition boundary mismatch can silently revert the planner to a full scan, and the first symptom is usually a trading partner reporting a timed-out verification call, not an internal alert.

Verification #

Prove the index earns its keep with a table-driven test that seeds a handful of events for one serial across two partitions and asserts the query returns the newest one:

import pytest
from datetime import datetime, timezone


@pytest.mark.asyncio
async def test_returns_latest_disposition(async_session):
    gtin, serial = "00312345000058", "SERIAL42"
    events = [
        ("commissioning", "active", datetime(2026, 6, 15, tzinfo=timezone.utc)),
        ("shipping", "in_transit", datetime(2026, 7, 2, tzinfo=timezone.utc)),
        ("decommissioning", "decommissioned", datetime(2026, 7, 10, tzinfo=timezone.utc)),
    ]
    for biz_step, disposition, event_time in events:
        async_session.add(
            EpcisEvent(
                gtin=gtin,
                serial_number=serial,
                event_time=event_time,
                event_type="ObjectEvent",
                biz_step=biz_step,
                disposition=disposition,
                read_point="urn:epc:id:sgln:0312345.00000.0",
            )
        )
    await async_session.commit()

    status = await get_current_status(async_session, gtin, serial)
    assert status is not None
    assert status["disposition"] == "decommissioned"
    assert status["event_time"] == events[-1][2]

Pair that correctness test with the explain_current_status helper from Step 5 run against a partition-sized fixture (tens of millions of synthetic rows, not a handful), and assert the returned plan text contains Index Only Scan and does not contain Seq Scan. A test that only checks the returned value can pass for years while the underlying plan silently degrades to a full scan as data volume grows — the plan assertion is what catches that regression before a trading partner does.

Gotchas & Edge Cases #

• Partition key must sit in the primary key. PostgreSQL requires every unique constraint on a partitioned table to include the partitioning column, which is why event_time joins gtin and serial_number in the primary key here rather than staying a plain indexed column.
• A missing time predicate defeats pruning. The composite index accelerates the GTIN-and-serial lookup regardless, but if the query never mentions event_time, the planner must still open every partition to check for matching rows. For point lookups this cost is small since each partition’s index makes the check fast, but for range queries always add a reasonable time bound.
• UTC vs. local event_time. Store and compare event_time as TIMESTAMPTZ, never a naive local timestamp — an ORDER BY event_time DESC over mixed offsets silently returns the wrong “latest” row, and the discrepancy only surfaces when a partner’s clock differs from yours.
• Leading zeros in GTIN. Keep gtin as a fixed-width VARCHAR(14), never a numeric type — casting to an integer strips leading zeros, breaks the GS1 modulo-10 check digit, and produces an index key that no longer matches the value a partner sends.
• Duplicate events at the same instant. Two events for the same unit can legitimately share an event_time to the millisecond during a high-speed line changeover; the primary key on (gtin, serial_number, event_time) will reject the second insert unless you widen the key or add a tiebreaker column, so decide that behavior deliberately rather than discovering it via a constraint violation in production.

Related #

• Up to the parent topic: EPCIS Repository Storage & Retention
• Enforcing Six-Year Retention with Python — rolling, detaching, and archiving the same monthly partitions this index design relies on
• Verification Router Service Architecture — the caller whose SLA this query’s latency has to satisfy
• Serialization Data Ingestion & EPCIS Event Sync — the broader pipeline this repository serves