perf(db): eliminate ResultRow wrapping overhead for PostgreSQL

hindsight-api-slim/hindsight_api/alembic/versions/k6l7m8n9o0p1_create_observation_sources_table.py

@@ -1,11 +1,10 @@
-"""Create observation_sources junction table
+"""No-op: observation_sources table is Oracle-only
 
-Replaces the source_memory_ids UUID[] column (PG) / CLOB (Oracle) with a
-proper junction table. This eliminates dialect-specific array operators
-(&&, unnest, JSON_TABLE) and enables standard SQL joins for all backends.
+Originally created the observation_sources junction table for all backends,
+but PG uses native array ops on the source_memory_ids column (faster at scale).
+Oracle creates this table via migrations_oracle.py instead.
 
-The old source_memory_ids column is retained for now (dual-write) and will
-be dropped in a future migration once all read paths are migrated.
+Kept as a no-op to preserve the Alembic revision chain.
 
 Revision ID: k6l7m8n9o0p1
 Revises: i4j5k6l7m8n9
@@ -14,57 +13,17 @@
 
 from collections.abc import Sequence
 
-from alembic import context, op
-
 revision: str = "k6l7m8n9o0p1"
 down_revision: str | Sequence[str] | None = "i4j5k6l7m8n9"
 branch_labels: str | Sequence[str] | None = None
 depends_on: str | Sequence[str] | None = None
 
 
-def _get_schema_prefix() -> str:
-    """Get schema prefix for table names (required for multi-tenant support)."""
-    schema = context.config.get_main_option("target_schema")
-    return f'"{schema}".' if schema else ""
-
-
 def upgrade() -> None:
-    schema = _get_schema_prefix()
-
-    # Create junction table.
-    # observation_id has ON DELETE CASCADE so deleting an observation cleans up its rows.
-    # source_id intentionally has NO FK — when a source memory is deleted, we need
-    # observation_sources rows to still exist so delete_stale_observations_for_memories()
-    # can find affected observations. Those observations are then deleted, which cascades
-    # to observation_sources via the observation_id FK.
-    op.execute(f"""
-        CREATE TABLE IF NOT EXISTS {schema}observation_sources (
-            observation_id UUID NOT NULL,
-            source_id UUID NOT NULL,
-            PRIMARY KEY (observation_id, source_id),
-            FOREIGN KEY (observation_id) REFERENCES {schema}memory_units(id) ON DELETE CASCADE
-        )
-    """)
-
-    # Index on source_id for reverse lookups (find observations referencing a given source)
-    op.execute(f"""
-        CREATE INDEX IF NOT EXISTS idx_obs_sources_source_id
-        ON {schema}observation_sources(source_id, observation_id)
-    """)
-
-    # Backfill from existing source_memory_ids array column
-    op.execute(f"""
-        INSERT INTO {schema}observation_sources (observation_id, source_id)
-        SELECT mu.id, unnest(mu.source_memory_ids)
-        FROM {schema}memory_units mu
-        WHERE mu.fact_type = 'observation'
-          AND mu.source_memory_ids IS NOT NULL
-          AND array_length(mu.source_memory_ids, 1) > 0
-        ON CONFLICT DO NOTHING
-    """)
+    # Oracle creates observation_sources via migrations_oracle.py.
+    # PG uses source_memory_ids array column directly — no junction table needed.
+    pass
 
 
 def downgrade() -> None:
-    schema = _get_schema_prefix()
-    op.execute(f"DROP INDEX IF EXISTS {schema}idx_obs_sources_source_id")
-    op.execute(f"DROP TABLE IF EXISTS {schema}observation_sources")
+    pass

hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py

@@ -1025,21 +1025,21 @@ async def _execute_update_action(
         merged_tags,
     )
 
-    # Dual-write: sync observation_sources junction table with updated source_ids.
-    # DELETE + INSERT is simpler than diffing, and this runs inside a transaction.
-    obs_uuid = uuid.UUID(observation_id)
-    await conn.execute(
-        f"DELETE FROM {fq_table('observation_sources')} WHERE observation_id = $1",
-        obs_uuid,
-    )
-    if source_ids:
-        await conn.executemany(
-            f"""
-            INSERT INTO {fq_table("observation_sources")} (observation_id, source_id)
-            VALUES ($1, $2)
-            """,
-            [(obs_uuid, sid) for sid in source_ids],
+    # Sync observation_sources junction table (Oracle only — PG uses native array ops).
+    if memory_engine._backend.ops.uses_observation_sources_table:
+        obs_uuid = uuid.UUID(observation_id)
+        await conn.execute(
+            f"DELETE FROM {fq_table('observation_sources')} WHERE observation_id = $1",
+            obs_uuid,
         )
+        if source_ids:
+            await conn.executemany(
+                f"""
+                INSERT INTO {fq_table("observation_sources")} (observation_id, source_id)
+                VALUES ($1, $2)
+                """,
+                [(obs_uuid, sid) for sid in source_ids],
+            )
 
     if perf:
         perf.record_timing("db_write", time.time() - t0)
@@ -1411,10 +1411,8 @@ async def _create_observation_directly(
         obs_mentioned_at,
     )
 
-    # Dual-write: populate observation_sources junction table alongside
-    # the source_memory_ids column. The junction table enables portable SQL
-    # joins, replacing PG-specific array operators and Oracle JSON_TABLE.
-    if source_memory_ids:
+    # Populate observation_sources junction table (Oracle only — PG uses native array ops).
+    if memory_engine._backend.ops.uses_observation_sources_table and source_memory_ids:
         await conn.executemany(
             f"""
             INSERT INTO {fq_table("observation_sources")} (observation_id, source_id)

hindsight-api-slim/hindsight_api/engine/db/ops.py

@@ -42,6 +42,15 @@ class DataAccessOps(ABC):
     in execution strategy between backends.
     """
 
+    @property
+    def uses_observation_sources_table(self) -> bool:
+        """Whether this backend uses the observation_sources junction table.
+
+        PG uses native array ops (source_memory_ids column) for reads and
+        skips junction table writes.  Oracle uses the junction table for both.
+        """
+        return True  # Default: use junction table (Oracle)
+
     # -- Bulk insert operations ------------------------------------------
 
     @abstractmethod
@@ -248,8 +257,8 @@ async def expand_observations(
     ) -> tuple[list[ResultRow], list[ResultRow], list[ResultRow]]:
         """Observation-specific graph expansion.
 
-        Both backends use the observation_sources junction table with standard
-        SQL joins. Previously PG used native array ops and Oracle used JSON_TABLE.
+        PG uses native array ops (source_memory_ids column) for performance.
+        Oracle uses the observation_sources junction table.
         """
         ...
 

hindsight-api-slim/hindsight_api/engine/db/ops_oracle.py

@@ -13,7 +13,7 @@
 
 from .base import DatabaseConnection
 from .ops import DataAccessOps, TagListingParts
-from .result import ResultRow
+from .result import DictResultRow as ResultRow
 
 
 class OracleOps(DataAccessOps):

hindsight-api-slim/hindsight_api/engine/db/ops_postgresql.py

@@ -17,6 +17,10 @@
 class PostgreSQLOps(DataAccessOps):
     """PostgreSQL-specific data access operations using unnest and LATERAL."""
 
+    @property
+    def uses_observation_sources_table(self) -> bool:
+        return False  # PG uses native array ops on source_memory_ids
+
     async def bulk_upsert_chunks(
         self,
         conn: DatabaseConnection,
@@ -450,23 +454,21 @@ async def expand_observations(
         budget: int,
         per_entity_limit: int,
     ) -> tuple[list[ResultRow], list[ResultRow], list[ResultRow]]:
-        # Entity expansion via observation_sources junction table.
-        # Previously used PG-specific unnest(source_memory_ids) and array
-        # overlap (&&). The junction table approach is portable across backends.
+        # v0.5.6 array ops: unnest, &&, COUNT(DISTINCT) on source_memory_ids.
         from ..schema import fq_table
 
-        obs_sources_table = fq_table("observation_sources")
         entity_rows = await conn.fetch(
             f"""
-            WITH source_ids AS (
-                SELECT DISTINCT os.source_id
-                FROM {obs_sources_table} os
-                WHERE os.observation_id = ANY($1::uuid[])
+            WITH seed_sources AS (
+                SELECT DISTINCT unnest(source_memory_ids) AS source_id
+                FROM {mu_table}
+                WHERE id = ANY($1::uuid[])
+                  AND source_memory_ids IS NOT NULL
             ),
             source_entities AS (
                 SELECT DISTINCT ue_seed.entity_id
-                FROM source_ids si
-                JOIN {ue_table} ue_seed ON ue_seed.unit_id = si.source_id
+                FROM seed_sources ss
+                JOIN {ue_table} ue_seed ON ue_seed.unit_id = ss.source_id
             ),
             connected_sources AS (
                 SELECT DISTINCT t.unit_id AS source_id
@@ -478,25 +480,23 @@ async def expand_observations(
                     ORDER BY ue_target.unit_id DESC
                     LIMIT {per_entity_limit}
                 ) t
-                WHERE t.unit_id NOT IN (SELECT source_id FROM source_ids)
+                WHERE NOT EXISTS (
+                    SELECT 1 FROM seed_sources ss WHERE ss.source_id = t.unit_id
+                )
+            ),
+            connected_array AS (
+                SELECT array_agg(source_id) AS source_ids FROM connected_sources
             )
             SELECT
                 mu.id, mu.text, mu.context, mu.event_date, mu.occurred_start,
                 mu.occurred_end, mu.mentioned_at,
                 mu.fact_type, mu.document_id, mu.chunk_id, mu.tags, mu.proof_count,
-                (SELECT COUNT(*)
-                 FROM {obs_sources_table} os2
-                 WHERE os2.observation_id = mu.id
-                   AND os2.source_id IN (SELECT source_id FROM connected_sources)
-                )::float AS score
-            FROM {mu_table} mu
+                (SELECT COUNT(DISTINCT s) FROM unnest(mu.source_memory_ids) s WHERE s = ANY(ca.source_ids))::float AS score
+            FROM {mu_table} mu, connected_array ca
             WHERE mu.fact_type = 'observation'
               AND mu.id != ALL($1::uuid[])
-              AND EXISTS (
-                  SELECT 1 FROM {obs_sources_table} os3
-                  WHERE os3.observation_id = mu.id
-                    AND os3.source_id IN (SELECT source_id FROM connected_sources)
-              )
+              AND ca.source_ids IS NOT NULL
+              AND mu.source_memory_ids && ca.source_ids
             ORDER BY score DESC
             LIMIT $2
             """,

hindsight-api-slim/hindsight_api/engine/db/oracle.py

@@ -35,7 +35,7 @@ def default(self, o):
 
 
 from .base import DatabaseBackend, DatabaseConnection
-from .result import ResultRow
+from .result import DictResultRow as ResultRow
 
 logger = logging.getLogger(__name__)
 

hindsight-api-slim/hindsight_api/engine/db/postgresql.py

@@ -1,7 +1,10 @@
 """PostgreSQL backend implementation using asyncpg.
 
 Wraps asyncpg's pool and connection objects behind the DatabaseBackend
-and DatabaseConnection interfaces.
+and DatabaseConnection interfaces.  Returns raw asyncpg.Record objects
+from fetch/fetchrow — they satisfy the ResultRow protocol natively in C,
+avoiding Python-level wrapping overhead (~570K __getitem__ calls per
+20-query benchmark → measurable regression when wrapped).
 """
 
 import logging
@@ -12,7 +15,6 @@
 import asyncpg  # noqa: F401
 
 from .base import DatabaseBackend, DatabaseConnection
-from .result import ResultRow
 
 logger = logging.getLogger(__name__)
 
@@ -36,15 +38,15 @@ async def execute(self, query: str, *args: Any, timeout: float | None = None) ->
     async def executemany(self, query: str, args: list[tuple[Any, ...]], *, timeout: float | None = None) -> None:
         await self._conn.executemany(query, args, timeout=timeout)
 
-    async def fetch(self, query: str, *args: Any, timeout: float | None = None) -> list[ResultRow]:
-        rows = await self._conn.fetch(query, *args, timeout=timeout)
-        return [ResultRow(row) for row in rows]
+    async def fetch(self, query: str, *args: Any, timeout: float | None = None) -> list:
+        # Return raw asyncpg.Record objects — they satisfy the ResultRow
+        # protocol natively (key access, .keys(), .get(), etc.) with zero
+        # Python wrapping overhead.
+        return await self._conn.fetch(query, *args, timeout=timeout)
 
-    async def fetchrow(self, query: str, *args: Any, timeout: float | None = None) -> ResultRow | None:
-        row = await self._conn.fetchrow(query, *args, timeout=timeout)
-        if row is None:
-            return None
-        return ResultRow(row)
+    async def fetchrow(self, query: str, *args: Any, timeout: float | None = None):
+        # Return raw asyncpg.Record — no wrapping needed.
+        return await self._conn.fetchrow(query, *args, timeout=timeout)
 
     async def fetchval(self, query: str, *args: Any, column: int = 0, timeout: float | None = None) -> Any:
         return await self._conn.fetchval(query, *args, column=column, timeout=timeout)