RFC-0013: High-Throughput Database Operations, Concurrency Controls, and Adapter Optimizations

• Status: Implemented
• Author(s): CalyXx
• Created: 2026-09-14
• Updated: 2026-09-21

---

1. Summary

This RFC specifies the architecture and implementation specifications for five critical high-throughput database primitives across libs/framework-m-core and libs/framework-m-standard:

1. Vectorized Multi-Row Insert (bulk_create): Single-statement multi-row SQL insertion bypassing per-row existence lookups, reducing database network round-trips from $O(N)$ to $O(1)$.
2. Bulk Field Mutation (bulk_set_value): Single-statement batch update for arbitrary identifier sets (WHERE name IN (...)), eliminating iterative single-document update queries.
3. Row-Level Concurrency Locking (FOR UPDATE SKIP LOCKED): Non-blocking worker queue claiming semantics in GenericRepository and query builders for horizontally scaled environments (Kubernetes pods, Celery/ARQ workers, outbox dispatchers), with transparent SQLite bypass for local test suites.
4. Automatic PgBouncer Detection: Dynamic detection of transaction-pooling proxies via URL port (6432) or PGBOUNCER_ENABLED flag, automatically configuring statement_cache_size=0 on asyncpg engines to prevent prepared statement collision errors.
5. Partial / Conditional Index DDL (IndexSpec with where): First-class support for partial indexes in DocType Meta.indexes definitions, producing dialect-optimized DDL (postgresql_where and sqlite_where) to dramatically minimize index storage and index write overhead for transactional queues and soft-deleted entities.

---

2. Motivation

Framework M targets zero-cliff scaling, transitioning smoothly from single-instance developer environments to hyperscale, multi-tenant distributed cloud infrastructure. However, real-world deployments running high-volume ingestion, message outbox dispatching, and distributed background processing have exposed four critical database adapter bottlenecks:

2.1 The $O(N)$ Round-Trip Query Storm in Bulk Ingestion

The existing repo.bulk_save(session, entities) API acts as an upsert mechanism. Before persisting, it iterates through all items and executes await self._entity_exists(session, entity):

# Current generic_repository.py: bulk_save()
for entity in entities:
    if await self._entity_exists(
        session, entity
    ):  # Executes: SELECT 1 FROM table WHERE id = ...
        existing_entities.append(entity)
    else:
        new_entities.append(entity)

When an outbox publisher or data ingestion pipeline imports 5,000 records, bulk_save generates 5,000 separate SELECT queries across the network before executing the insert. In workloads where records are guaranteed to be novel (e.g., event outbox logs, telemetry, audit entries, ledger rows), this causes catastrophic database latency and connection pool exhaustion.

Similarly, updating a field across a list of documents requires iterative calls to db.set_value(...), executing $N$ distinct UPDATE statements instead of a single batched query.

2.2 Worker Race Conditions and Contention in Multi-Pod Scaling

When background workers (outbox dispatchers, webhook relays, scheduled job processors) scale horizontally across multiple pods, they compete for pending tasks:

-- Worker 1 & Worker 2 simultaneously execute:
SELECT * FROM tabOutboxEntry WHERE status = 'Pending' ORDER BY creation ASC LIMIT 50;

Without row-level locking, both pods select the same batch, causing duplicate job execution and downstream side effects. If standard FOR UPDATE is used, Worker 2 blocks completely until Worker 1 finishes its transaction, destroying parallelism and inducing serialization deadlocks. Postgres supports FOR UPDATE SKIP LOCKED, allowing workers to immediately skip rows locked by peers and process subsequent tasks. Framework M currently lacks a repository abstraction for SKIP LOCKED.

2.3 PgBouncer Prepared Statement Cache Collisions

In Kubernetes deployments, Postgres instances sit behind connection poolers like PgBouncer in transaction pooling mode. By default, asyncpg aggressively prepares statements on the connection and caches statement signatures (statement_cache_size=100). In transaction pooling, sequential transactions from the same client connection land on different backend PostgreSQL server processes. When asyncpg executes a cached statement name, the backend throws:

asyncpg.exceptions.InvalidSQLStatementNameError: prepared statement "_pg3_0" does not exist

Requiring developers to manually configure connection strings with ?prepared_statement_cache_size=0 is fragile and frequently missed in staging/production promotions.

2.4 Index Bloat on Status-Driven Tables

In outbox patterns, task queues, and soft-delete workflows, 99.9% of rows reside in a terminal state (Completed, Cancelled), while only 0.1% are actionable (Pending, Processing). A standard B-Tree index on (status, next_attempt_at) indexes all 50,000,000 rows, consuming gigabytes of RAM and imposing write overhead on every update. A partial index (WHERE status = 'Pending') indexes only the active 500 rows, consuming mere kilobytes and fitting entirely into CPU L1/L2 cache. Framework M's SchemaMapper currently only builds full-table indexes.

---

3. High-Level Architecture & Component Interaction

The following diagram illustrates how these features integrate into the core repository, query pipeline, connection factory, and schema generation:

┌─────────────────────────────────────────────────────────────────────────────────────────┐
│                                   APPLICATION / CALLER                                  │
│   (Controllers, Outbox Dispatchers, Ingestion Pipelines, Background Task Workers)       │
└───────────────────────────┬─────────────────────────────────┬───────────────────────────┘
                            │                                 │
           await repo.bulk_create(entities)    await repo.list_entities(for_update=True,
           await db.bulk_set_value(names, vals)                         skip_locked=True)
                            │                                 │
                            ▼                                 ▼
┌─────────────────────────────────────────────────────────────────────────────────────────┐
│                           FRAMEWORK-M-STANDARD / ADAPTERS / DB                          │
│                                                                                         │
│  ┌───────────────────────────┐    ┌───────────────────────────┐   ┌──────────────────┐  │
│  │    ControllerDbAdapter    │    │     GenericRepository     │   │   QueryBuilder   │  │
│  │ - bulk_insert()           │───▶│ - bulk_create()           │──▶│ - apply_filters  │  │
│  │ - bulk_set_value()        │    │ - bulk_set_value()        │   │ - apply_ordering │  │
│  │                           │    │ - list_entities()         │   │ - apply_locking  │  │
│  └───────────────────────────┘    └─────────────┬─────────────┘   └────────┬─────────┘  │
│                                                 │                          │            │
│               ┌─────────────────────────────────┴──────────────────────────┘            │
│               │ (Dialect Inspection: PostgreSQL vs SQLite)                              │
│               ▼                                                                         │
│  ┌───────────────────────────────────────────────┐   ┌───────────────────────────────┐  │
│  │              ConnectionFactory                │   │         SchemaMapper          │  │
│  │ - Detect PgBouncer (Port 6432 / Env Flag)     │   │ - IndexSpec with where clause │  │
│  │ - Injects asyncpg statement_cache_size=0      │   │ - postgresql_where / sqlite   │  │
│  └───────────────────────┬───────────────────────┘   └───────────────┬───────────────┘  │
└──────────────────────────┼───────────────────────────────────────────┼──────────────────┘
                           │                                           │
                           ▼                                           ▼
┌──────────────────────────────────────────────────┐   ┌──────────────────────────────────┐
│             DATABASE ENGINE / INFRA              │   │         GENERATED DDL            │
│  - PostgreSQL + PgBouncer (Port 6432)            │   │ CREATE INDEX idx_outbox_due ON   │
│  - Multi-row INSERT INTO ... VALUES (...), (...) │   │ tabOutboxEntry (status, attempt) │
│  - SELECT ... FOR UPDATE SKIP LOCKED             │   │ WHERE status = 'Pending';        │
└──────────────────────────────────────────────────┘   └──────────────────────────────────┘

---

4. Detailed Design

4.1 Vectorized Multi-Row Insert (bulk_create)

4.1.1 Core Interface Definition (libs/framework-m-core)

We extend RepositoryProtocol[T] in framework_m_core/interfaces/repository.py to add bulk_create:

class RepositoryProtocol[T](Protocol):
    ...

    async def bulk_create(self, entities: Sequence[T]) -> Sequence[T]:
        """Insert multiple entities in a single vectorized SQL INSERT statement.

        Unlike bulk_save(), bulk_create() assumes all entities are new and
        bypasses existence verification queries, providing maximum throughput.

        Args:
            entities: Sequence of entity instances to insert.

        Returns:
            Sequence of persisted entities with generated primary keys and timestamps.

        Raises:
            IntegrityError: If primary key or unique constraint violations occur.
            RepositoryError: On database execution failure.
        """
        ...

4.1.2 GenericRepository Implementation (libs/framework-m-standard)

In GenericRepository, bulk_create supports both session-explicit execution (within a UnitOfWork) and session-injected execution (via _session_factory).

async def bulk_create(
    self,
    session: AsyncSession | Sequence[T] = None,
    entities: Sequence[T] | None = None,
) -> list[T]:
    """Insert multiple entities in a single multi-row SQL statement."""
    # Handle polymorphic call signature (session-first vs entities-first)
    if isinstance(session, (list, tuple, Sequence)) and entities is None:
        entities = session
        session = None

    if not entities:
        return []

    sf = getattr(self, "_session_factory", None)
    if session is None and sf is not None:
        async with sf.get_session() as auto_session:
            return await self.bulk_create(auto_session, entities)

    if session is None:
        raise ValueError(
            "Active session or session_factory is required for bulk_create"
        )

    entity_list = list(entities)
    exclude_fields = self._mapper.get_excluded_fields()
    now = datetime.now(UTC)

    # 1. Deterministic batch identifier and timestamp population
    values_list: list[dict[str, Any]] = []
    for entity in entity_list:
        # Generate primary key / name if not provided
        if getattr(entity, "id", None) is None and hasattr(self._table.c, "id"):
            entity.id = uuid4()

        if getattr(entity, "name", None) is None and hasattr(self._table.c, "name"):
            entity.name = await self._naming_service.generate_name(session, entity)

        # Ensure consistent creation and modification timestamps
        if hasattr(entity, "creation") and entity.creation is None:
            entity.creation = now
        if hasattr(entity, "modified"):
            entity.modified = now

        row_data = self._mapper.entity_to_db_payload(entity, exclude_fields)
        row_data["modified"] = now
        if "creation" in self._table.c and "creation" not in row_data:
            row_data["creation"] = now

        values_list.append(row_data)

    # 2. Cryptographic hash chaining for tamper-proof DocTypes
    await self._tamper_manager.prepare_bulk_insert_hashes(
        session, entity_list, values_list
    )

    # 3. Multi-row vectorized SQL execution
    try:
        stmt = sa.insert(self._table)
        await session.execute(stmt, values_list)
        logger.debug(
            "Vectorized bulk insert: %d %s entities",
            len(entity_list),
            self._model.__name__,
        )
    except SAIntegrityError as e:
        error_msg = str(e.orig) if e.orig else str(e)
        logger.error(
            "Vectorized bulk insert failed for %s: %s",
            self._model.__name__,
            error_msg,
        )
        raise IntegrityError(
            f"Bulk insert failed for {self._model.__name__}: {error_msg}"
        ) from e
    except SQLAlchemyError as e:
        logger.error("Database error during bulk insert: %s", e)
        raise DatabaseError(f"Database error during bulk insert: {e}") from e

    return entity_list

4.1.3 Delegation in ControllerDbAdapter

class ControllerDbAdapter:
    ...

    async def bulk_insert(self, doctype_cls: type[T], entities: list[T]) -> list[T]:
        """Insert multiple entities in a single vectorized SQL operation."""
        repo = self._get_repo(doctype_cls)
        return await repo.bulk_create(self._session, entities)

---

4.2 Bulk Field Mutation (bulk_set_value)

4.2.1 Repository & Controller Adapter Signatures

We introduce bulk_set_value to ControllerDbAdapter and GenericRepository.

In ControllerDbAdapter:

async def bulk_set_value(
    self,
    doctype_cls: type[T],
    names: list[str] | list[UUID],
    field_values: dict[str, Any],
) -> int:
    """Update field values for multiple documents matching names/IDs.

    Args:
        doctype_cls: The DocType class to update.
        names: List of document names or UUIDs.
        field_values: Key-value mapping of fields to update.

    Returns:
        The number of rows matched and updated.
    """
    if not names or not field_values:
        return 0

    repo = self._get_repo(doctype_cls)
    table = repo._table

    # Determine identifier column (id vs name)
    first_key = names[0]
    is_uuid = False
    if isinstance(first_key, UUID):
        is_uuid = True
    else:
        try:
            UUID(str(first_key))
            is_uuid = True
        except (ValueError, AttributeError):
            is_uuid = False

    target_col = None
    if is_uuid and hasattr(table.c, "id"):
        target_col = table.c.id
    elif hasattr(table.c, "name"):
        target_col = table.c.name
    elif hasattr(table.c, "id"):
        target_col = table.c.id
    else:
        raise RepositoryError(
            f"DocType {doctype_cls.__name__} has no name or id column"
        )

    # Automatically touch modified timestamp if present
    payload = dict(field_values)
    if "modified" not in payload and hasattr(table.c, "modified"):
        payload["modified"] = datetime.now(UTC)

    statement = sa.update(table).where(target_col.in_(names)).values(**payload)

    result = await self._session.execute(statement)
    return result.rowcount or 0

---

4.3 Row-Level Concurrency Locking (FOR UPDATE SKIP LOCKED)

4.3.1 Interface Update (RepositoryProtocol.list)

In libs/framework-m-core/src/framework_m_core/interfaces/repository.py:

class RepositoryProtocol[T](Protocol):
    ...

    async def list(
        self,
        filters: list[FilterSpec] | None = None,
        order_by: list[OrderSpec] | None = None,
        limit: int = 20,
        offset: int = 0,
        as_of: datetime | None = None,
        for_update: bool = False,
        skip_locked: bool = False,
    ) -> PaginatedResult[T]:
        """List entities with filtering, sorting, pagination, and optional row locks.

        Args:
            filters: Optional list of filter conditions.
            order_by: Optional list of sort specifications.
            limit: Maximum number of items to return.
            offset: Number of items to skip.
            as_of: Optional timestamp for temporal query.
            for_update: If True, locks selected rows for update (Postgres/MySQL).
            skip_locked: If True and for_update is True, skips locked rows.
        """
        ...

4.3.2 QueryBuilder Locking Logic (query_builder.py)

In libs/framework-m-standard/src/framework_m_standard/adapters/db/query_builder.py:

@staticmethod
def apply_locking(
    stmt: Any,
    dialect_name: str,
    for_update: bool = False,
    skip_locked: bool = False,
    nowait: bool = False,
) -> Any:
    """Apply row-level locking clauses with dialect safety.

    Args:
        stmt: SQLAlchemy Select statement.
        dialect_name: The name of the active SQL dialect (e.g. 'postgresql', 'sqlite').
        for_update: Whether to request an exclusive row lock.
        skip_locked: Whether to skip locked rows (PostgreSQL/MySQL 8+ only).
        nowait: Whether to fail immediately if rows are locked.

    Returns:
        The modified statement.
    """
    if not for_update:
        return stmt

    # SQLite does not support row-level locks; bypass safely
    if dialect_name == "sqlite":
        logger.debug("Bypassing with_for_update on SQLite dialect")
        return stmt

    # PostgreSQL, MySQL 8+, Oracle support SKIP LOCKED
    return stmt.with_for_update(skip_locked=skip_locked, nowait=nowait)

4.3.3 Integration in GenericRepository.list_entities

In libs/framework-m-standard/src/framework_m_standard/adapters/db/generic_repository.py:

async def list_entities(
    self,
    session: AsyncSession | Any = None,
    filters: list[FilterSpec] | None = None,
    order_by: list[OrderSpec] | None = None,
    limit: int = 20,
    offset: int = 0,
    as_of: datetime | None = None,
    include_deleted: bool = False,
    count_threshold: int = 0,
    for_update: bool = False,
    skip_locked: bool = False,
) -> PaginatedResult[T]:
    ...
    # Apply ordering and pagination
    stmt = stmt.limit(effective_limit).offset(offset)

    # Apply row-level locking
    if for_update:
        bind = (
            session.get_bind()
            if hasattr(session, "get_bind")
            else getattr(session, "bind", None)
        )
        dialect_name = bind.dialect.name if bind else ""
        stmt = QueryBuilder.apply_locking(
            stmt,
            dialect_name=dialect_name,
            for_update=True,
            skip_locked=skip_locked,
        )

    result = await session.execute(stmt)
    rows = result.fetchall()
    ...

---

4.4 Automatic PgBouncer Connection Detection

4.4.1 Mechanics in ConnectionFactory (connection.py)

PgBouncer operates primarily on port 6432. In libs/framework-m-standard/src/framework_m_standard/adapters/db/connection.py, we introduce automatic detection and configuration of connect_args:

from urllib.parse import urlparse


class ConnectionFactory:
    ...

    @staticmethod
    def _is_pgbouncer_target(url: str) -> bool:
        """Determine if target database URL connects through PgBouncer."""
        # 1. Check explicit environment flag
        env_flag = os.environ.get("PGBOUNCER_ENABLED", "").strip().lower()
        if env_flag in ("true", "1", "yes", "on"):
            return True

        # 2. Check standard PgBouncer port in URL
        try:
            parsed = urlparse(url)
            if parsed.port == 6432:
                return True
        except Exception:
            pass

        return False

    def _build_engine_kwargs(
        self,
        expanded_url: str,
        base_kwargs: dict[str, Any],
    ) -> dict[str, Any]:
        """Construct engine parameters with PgBouncer protections."""
        kwargs = dict(base_kwargs)

        if expanded_url.startswith("postgresql"):
            connect_args = kwargs.setdefault("connect_args", {})
            if self._is_pgbouncer_target(expanded_url):
                logger.info(
                    "PgBouncer detected on %s; setting statement_cache_size=0 on asyncpg engine",
                    expanded_url.split("@")[
                        -1
                    ],  # Sanitized logging without credentials
                )
                # Disable asyncpg prepared statement caching for transaction pooling
                connect_args["statement_cache_size"] = 0
                connect_args["prepared_statement_cache_size"] = 0

        return kwargs

This is invoked inside both configure() and get_engine_async() when initializing create_async_engine.

---

4.5 Partial / Conditional Indexes (IndexSpec with where)

4.5.1 Schema Models (framework_m_core/domain/meta.py)

We update IndexDef in libs/framework-m-core/src/framework_m_core/domain/meta.py to include where: str | None = None and export IndexSpec = IndexDef:

class IndexDef(BaseModel):
    """Strict definition for a database index."""

    fields: list[str]
    unique: bool = False
    name: str | None = None
    where: str | None = None  # SQL WHERE predicate for partial / conditional indexes

    model_config = ConfigDict(extra="allow")


# Public alias for developer ergonomics
IndexSpec = IndexDef

4.5.2 Schema Mapper DDL Generation (schema_mapper.py)

In libs/framework-m-standard/src/framework_m_standard/adapters/db/schema_mapper.py, _add_custom_indexes is enhanced to parse the where attribute and supply it to SQLAlchemy's dialect-specific parameters:

def _add_custom_indexes(self, table: Table, model: type[BaseDocType]) -> None:
    from sqlalchemy import Index, text

    indexes = model.get_indexes()
    for index_spec in indexes:
        fields: list[str] = []
        where_clause: str | None = None
        index_name: str | None = None
        is_unique: bool = False

        if isinstance(index_spec, dict):
            fields = index_spec.get("fields", [])
            where_clause = index_spec.get("where")
            index_name = index_spec.get("name")
            is_unique = index_spec.get("unique", False)
        elif hasattr(index_spec, "fields"):  # IndexDef / IndexSpec instance
            fields = index_spec.fields
            where_clause = getattr(index_spec, "where", None)
            index_name = getattr(index_spec, "name", None)
            is_unique = getattr(index_spec, "unique", False)
        elif isinstance(index_spec, (list, tuple)):
            fields = list(index_spec)

        if not fields:
            continue

        columns = [table.c[f] for f in fields if f in table.c]
        if not columns:
            continue

        if not index_name:
            field_names = "_".join(fields)
            index_name = f"ix_{table.name}_{field_names}"

        index_kwargs: dict[str, Any] = {"unique": is_unique}
        if where_clause:
            # PostgreSQL and SQLite support partial indexes via where predicates
            index_kwargs["postgresql_where"] = text(where_clause)
            index_kwargs["sqlite_where"] = text(where_clause)

        Index(index_name, *columns, **index_kwargs)

4.5.3 Introspection Protocol Support (introspection.py)

In libs/framework-m-core/src/framework_m_core/interfaces/introspection.py:

class IndexMeta(BaseModel):
    """Normalized metadata for a database index or constraint."""

    name: str
    columns: list[str]
    unique: bool = False
    where: str | None = None  # Normalized predicate for schema diffing

This ensures that schema change detection (detect_schema_changes) detects modifications to partial index predicates without triggering spurious table recreations.

---

5. Database Dialect Compatibility Matrix

Feature	PostgreSQL 14+	SQLite 3.35+	MySQL 8.0+	MariaDB 10.6+
Multi-row INSERT (bulk_create)	Full (VALUES (...), (...))	Full	Full	Full
bulk_set_value (WHERE name IN (...))	Full	Full	Full	Full
FOR UPDATE SKIP LOCKED	Full (FOR UPDATE SKIP LOCKED)	Bypassed (Table lock fallback)	Full (SKIP LOCKED)	Full (SKIP LOCKED)
PgBouncer Detection (statement_cache_size=0)	Full (Target system)	N/A (Ignored)	N/A (Ignored)	N/A (Ignored)
Partial Index (WHERE <predicate>)	Full (CREATE INDEX ... WHERE)	Full (SQLite 3.8+)	N/A (Omitted or Functional)	N/A (Ignored)

---

6. Security, Tenancy & Transactional Integrity

1. Row-Level Security (RLS):
   • bulk_create automatically enforces the tenant's context (owner and tenant_id fields) derived from UserContext.
   • bulk_set_value in controllers runs within the active session boundaries. Where tenant isolation applies, developers must pass tenant-scoped identifiers or include tenant_id in the filter spec.
2. Optimistic Concurrency Control (OCC):
   • bulk_create applies only to new records, where versioning initializes at version = 1.
   • bulk_set_value is intended for administrative or worker state transitions (e.g. status updates). OCC verification is bypassed for batch efficiency; if OCC guarantees are strictly required per-entity, callers should use bulk_save() or single-document save().
3. Tamper-Proof Chaining:
   • bulk_create maintains compatibility with Framework M's cryptographic audit logs by delegating to _tamper_manager.prepare_bulk_insert_hashes(...) to compute sequential SHA-256 rolling hashes across the batch before execution.

---

7. Drawbacks

• No Lifecycle Hooks on Vectorized Bulk Operations:
  • bulk_create deliberately bypasses per-entity controller hooks (before_insert, after_insert) to achieve $O(1)$ vectorized performance. If complex entity validation or side-effects are required per-document, callers must continue using save().
• Dialect Disparities on Partial Indexes:
  • Partial indexes are unsupported on MySQL/MariaDB. Developers writing cross-database applications must be aware that conditional index predicates are ignored on non-Postgres/SQLite engines.

---

8. Alternatives Considered

Alternative	Pros	Cons	Reason for Rejection
Keep bulk_save Only	No new API surface	Incurs $O(N)$ existence queries; crippling performance on high-volume inserts	Incompatible with high-throughput outbox and ingestion targets
Client-Side Queueing without SKIP LOCKED	Works on all databases including SQLite	Requires centralized Redis/RabbitMQ coordinator; introduces extra infra dependency	Violates Framework M's "zero-cliff" philosophy of native database primitives
Manual URL Flags for PgBouncer	Zero framework code changes	Developers frequently forget query parameters; breaks in production K8s	Silent failure in production transaction pooling clusters
Application-Level Filtering over Full Indexes	Portable across all SQL dialects	Wastes massive disk storage and degrades write performance on 99% inactive rows	Inefficient for outbox tables containing millions of historic rows

---

9. Migration & Backwards Compatibility

• Breaking Changes: None. All new methods (bulk_create, bulk_set_value) are additive.
• bulk_save Compatibility: Existing calls to repo.bulk_save() continue to function identically with upsert semantics.
• Existing Meta.indexes: Retains full backwards compatibility with tuple, list, and dict index declarations; where and IndexSpec are strictly optional additions.

---

10. Unresolved Questions

1. Batch Chunking Thresholds:
   • Should bulk_create automatically chunk batches exceeding a specific threshold (e.g., 2,000 rows) to stay within parameter limits on engines with tight binding constraints (like SQLite's 32,766 parameter limit)?
   • Proposed Decision: Default to chunking at 1,000 rows for SQLite, while allowing unbounded or configurable batch sizes on PostgreSQL.

---

11. Implementation Plan

Phase 1: Core Definitions & Spec

• Add IndexSpec = IndexDef with where: str | None = None in framework_m_core.domain.meta.
• Add where: str | None = None to IndexMeta in framework_m_core.interfaces.introspection.
• Add bulk_create and locking parameters (for_update, skip_locked) to RepositoryProtocol in framework_m_core.interfaces.repository.

Phase 2: High-Performance Generic Repository

• Implement bulk_create in GenericRepository (generic_repository.py).
• Implement apply_locking with SQLite bypass in QueryBuilder (query_builder.py).
• Wire for_update and skip_locked into GenericRepository.list_entities.

Phase 3: Controller DB Adapter

• Implement bulk_set_value and bulk_insert on ControllerDbAdapter in generic_repository.py.

Phase 4: PgBouncer Auto-Detection

• Implement _is_pgbouncer_target and automatic statement_cache_size=0 in ConnectionFactory (connection.py).

Phase 5: Schema Mapper DDL & Migration Diffing

• Update SchemaMapper._add_custom_indexes to generate postgresql_where and sqlite_where.
• Update schema diff engine to detect partial index changes.

Phase 6: Test Suite & Verification

• Unit tests for bulk_create confirming single SQL statement execution (test_generic_repository_bulk.py).
• Unit tests for bulk_set_value on ControllerDbAdapter (test_controller_db_bulk.py).
• Multi-worker concurrency tests with FOR UPDATE SKIP LOCKED (test_skip_locked.py).
• PgBouncer detection unit tests (test_pgbouncer_detection.py).
• Partial index DDL tests in test_partial_index.py.

---

12. References

• PostgreSQL Documentation: FOR UPDATE SKIP LOCKED
• PostgreSQL Documentation: Partial Indexes
• PgBouncer Documentation: Transaction Pooling and Prepared Statements
• SQLAlchemy Documentation: Index.postgresql_where
• Framework M Roadmap