test(tables): isolate the fence plan fixture and require both racers to wait

Author: waleedlatif1

apps/sim/lib/table/rows/row-writes.integration.ts

@@ -12,8 +12,9 @@ import { tableBillingMock, tableBillingMockFns } from '@sim/testing/mocks/table-
 import { tableTriggerMock } from '@sim/testing/mocks/table-trigger.mock'
 import { tableWorkflowColumnsMock } from '@sim/testing/mocks/table-workflow-columns.mock'
 import { sleep } from '@sim/utils/helpers'
-import { generateId } from '@sim/utils/id'
+import { generateId, generateShortId } from '@sim/utils/id'
 import { sql } from 'drizzle-orm'
+import { drizzle } from 'drizzle-orm/postgres-js'
 import postgres from 'postgres'
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
@@ -23,7 +24,6 @@ vi.mock('@/lib/table/workflow-columns', () => tableWorkflowColumnsMock)
 vi.mock('@/lib/uploads/core/storage-service', () => storageServiceMock)
 
 import { USER_TABLE_ROWS_SQL_NAME } from '@/lib/table/constants'
-import { withSeqscanOff } from '@/lib/table/planner'
 import {
   batchInsertRows,
   deleteRowsByFilter,
@@ -45,15 +45,10 @@ const control = postgres(url, { max: 4, onnotice: () => {} })
 const workspaceId = generateId()
 const userId = generateId()
 
-/**
- * `db:push` never installs migration-only objects: the `rows_version` triggers and the
- * `(workspace_id, table_id)` dependency statistic. Only the cases that depend on them need them.
- */
-const [{ migrated }] = await control<{ migrated: boolean }[]>`SELECT
-  EXISTS (SELECT 1 FROM pg_trigger WHERE tgname = 'user_table_rows_version_update_trigger')
-  AND EXISTS (
-    SELECT 1 FROM pg_statistic_ext WHERE stxname = 'user_table_rows_workspace_table_stats'
-  ) AS migrated`
+/** `db:push` never installs the migration-only `rows_version` triggers; only those cases need them. */
+const [{ migrated }] = await control<{ migrated: boolean }[]>`SELECT EXISTS (
+  SELECT 1 FROM pg_trigger WHERE tgname = 'user_table_rows_version_update_trigger'
+) AS migrated`
 
 async function createTable(columns: ColumnDefinition[]): Promise<TableDefinition> {
   const id = generateId()
@@ -80,6 +75,7 @@ async function rowsVersion(tableId: string): Promise<number> {
 interface QueryPlan {
   'Node Type': string
   'Index Name'?: string
+  'Subplan Name'?: string
   Plans?: QueryPlan[]
 }
 
@@ -383,36 +379,74 @@ describe('table row writes against real PostgreSQL', () => {
 
       expect([...result.affectedRowIds].sort()).toEqual([...expected].sort())
     })
+  })
 
-    /** A table large enough that the planner's index choice reflects its statistics. */
-    async function seedPlanFixture(): Promise<(filter: Filter) => Promise<QueryPlan[]>> {
-      const table = await createTable(textColumns('kind', 'n'))
-      await control`INSERT INTO user_table_rows (id, table_id, workspace_id, data, order_key)
-        SELECT ${table.id} || '-' || n, ${table.id}, ${workspaceId},
+  /**
+   * Plans come from a private copy of `user_table_rows` (same columns and indexes, no triggers) in
+   * its own schema, holding only this fixture and analyzed from a full sample. Integration files
+   * share one database in parallel, so plans against the shared relation would depend on every
+   * other suite's rows and statistics.
+   */
+  describe('limited filtered query plans', () => {
+    const planSchema = `row_writes_${generateShortId()
+      .replace(/[^a-zA-Z0-9]/g, '')
+      .toLowerCase()}`
+    const planConnection = postgres(url, {
+      max: 1,
+      onnotice: () => {},
+      connection: { search_path: `${planSchema}, public` },
+    })
+    const planTable: TableDefinition = {
+      id: 'table-plan',
+      workspaceId: 'workspace-plan',
+      name: 'Plan fixture',
+      description: null,
+      schema: { columns: textColumns('kind', 'n') },
+      metadata: null,
+      rowCount: 20_000,
+      maxRows: 100_000,
+      createdBy: 'user-plan',
+      locks: { schemaLocked: false, insertLocked: false, updateLocked: false, deleteLocked: false },
+      archivedAt: null,
+      createdAt: new Date(0),
+      updatedAt: new Date(0),
+    }
+
+    beforeAll(async () => {
+      await planConnection`CREATE SCHEMA ${planConnection(planSchema)}`
+      await planConnection`CREATE TABLE user_table_rows
+        (LIKE public.user_table_rows INCLUDING DEFAULTS INCLUDING INDEXES)`
+      await planConnection`INSERT INTO user_table_rows (id, table_id, workspace_id, data, order_key)
+        SELECT 'row-' || n, ${planTable.id}, ${planTable.workspaceId},
           jsonb_build_object('kind', CASE WHEN n % 500 = 0 THEN 'hit' ELSE 'miss' END, 'n', n::text),
           'a' || lpad(n::text, 6, '0')
         FROM generate_series(1, 20000) AS n`
-      await control`ANALYZE user_table_rows`
+      await planConnection`ANALYZE user_table_rows`
+    })
 
-      return async (filter) => {
-        const filterClause = buildFilterClause(
-          filter,
-          USER_TABLE_ROWS_SQL_NAME,
-          table.schema.columns
-        )
-        if (!filterClause) throw new Error('Fixture filter compiled to no clause')
-        const plans = await withSeqscanOff((trx) =>
-          trx.execute<{ 'QUERY PLAN': { Plan: QueryPlan }[] }>(
-            sql`EXPLAIN (FORMAT JSON) ${firstMatchingRowsQuery(table, filter, filterClause, 5, 'rows')}`
-          )
+    afterAll(async () => {
+      await planConnection`DROP SCHEMA ${planConnection(planSchema)} CASCADE`
+      await planConnection.end()
+    })
+
+    async function planFor(filter: Filter): Promise<QueryPlan[]> {
+      const filterClause = buildFilterClause(
+        filter,
+        USER_TABLE_ROWS_SQL_NAME,
+        planTable.schema.columns
+      )
+      if (!filterClause) throw new Error('Fixture filter compiled to no clause')
+      const query = firstMatchingRowsQuery(planTable, filter, filterClause, 5, 'rows')
+      const plans = await drizzle(planConnection).transaction(async (trx) => {
+        await trx.execute(sql`SET LOCAL enable_seqscan = off`)
+        return trx.execute<{ 'QUERY PLAN': { Plan: QueryPlan }[] }>(
+          sql`EXPLAIN (FORMAT JSON) ${query}`
         )
-        return planNodes(plans[0]['QUERY PLAN'][0].Plan)
-      }
+      })
+      return planNodes(plans[0]['QUERY PLAN'][0].Plan)
     }
 
     it('fences only a containment filter and leaves any other filter unfenced', async () => {
-      const planFor = await seedPlanFixture()
-
       const fenced = await planFor({ kind: 'hit' })
       expect(fenced.some((node) => node['Node Type'] === 'CTE Scan')).toBe(true)
 
@@ -421,17 +455,16 @@ describe('table row writes against real PostgreSQL', () => {
       expect(walked[0]['Node Type']).toBe('Limit')
     })
 
-    it.skipIf(!migrated)('probes the tenant GIN index inside the fence', async () => {
-      const planFor = await seedPlanFixture()
-
-      const fenced = await planFor({ kind: 'hit' })
-      expect(
-        fenced.some(
-          (node) =>
-            node['Node Type'] === 'Bitmap Index Scan' &&
-            node['Index Name'] === 'user_table_rows_tenant_data_gin_idx'
-        )
-      ).toBe(true)
+    it('probes the containment GIN index inside the fence', async () => {
+      const matched = (await planFor({ kind: 'hit' })).find(
+        (node) => node['Subplan Name'] === 'CTE matched'
+      )
+      const probe =
+        matched && planNodes(matched).find((node) => node['Node Type'] === 'Bitmap Index Scan')
+      const [index] = await planConnection<{ indexdef: string }[]>`SELECT indexdef FROM pg_indexes
+        WHERE schemaname = ${planSchema} AND indexname = ${probe?.['Index Name'] ?? ''}`
+      expect(index?.indexdef).toContain('USING gin')
+      expect(index?.indexdef).toContain('jsonb_path_ops')
     })
   })
 
@@ -442,7 +475,8 @@ describe('table row writes against real PostgreSQL', () => {
 
     /**
      * Holds the table's row-order lock while both inserts start, so each one's pre-insert work runs
-     * before either can write. Released once both wait on the lock.
+     * before either can write. Released only once both are seen waiting on this table's lock;
+     * otherwise the inserts could run one after the other and prove nothing about the race.
      */
     async function raceUnderHeldOrderLock(
       tableId: string,
@@ -453,13 +487,18 @@ describe('table row writes against real PostgreSQL', () => {
         await holder`BEGIN`
         await holder`SELECT pg_advisory_xact_lock(hashtextextended(${`user_table_rows_pos:${tableId}`}, 0))`
         const racers = Promise.allSettled([insert(), insert()])
-        for (let attempt = 0; attempt < 400; attempt++) {
-          const [{ waiting }] = await control<{ waiting: number }[]>`SELECT count(*)::int AS waiting
-            FROM pg_locks WHERE locktype = 'advisory' AND NOT granted
-              AND database = (SELECT oid FROM pg_database WHERE datname = current_database())`
-          if (waiting >= 2) break
+        let waiting = 0
+        for (let attempt = 0; attempt < 400 && waiting < 2; attempt++) {
           await sleep(5)
+          ;[{ waiting }] = await control<{ waiting: number }[]>`
+            WITH lock AS (SELECT hashtextextended(${`user_table_rows_pos:${tableId}`}, 0) AS key)
+            SELECT count(*)::int AS waiting FROM pg_locks, lock
+            WHERE locktype = 'advisory' AND NOT granted AND objsubid = 1
+              AND database = (SELECT oid FROM pg_database WHERE datname = current_database())
+              AND classid = ((lock.key >> 32) & 4294967295)::oid
+              AND objid = (lock.key & 4294967295)::oid`
         }
+        expect(waiting).toBe(2)
         await holder`COMMIT`
         return await racers
       } finally {