improvement(tables): fence only containment filters and trim rows version trigger cost

Author: waleedlatif1

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

@@ -12,6 +12,7 @@ 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 { sql } from 'drizzle-orm'
 import postgres from 'postgres'
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
@@ -20,16 +21,20 @@ vi.mock('@/lib/table/trigger', () => tableTriggerMock)
 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,
+  firstMatchingRowsQuery,
   insertRow,
   updateRow,
   updateRowsByFilter,
 } from '@/lib/table/rows/service'
 import { getTableById } from '@/lib/table/service'
 import { getOrCreateTableSnapshot } from '@/lib/table/snapshot-cache'
-import type { ColumnDefinition, TableDefinition } from '@/lib/table/types'
+import { buildFilterClause } from '@/lib/table/sql'
+import type { ColumnDefinition, Filter, TableDefinition } from '@/lib/table/types'
 
 const url = readTestDatabaseUrl()
 if (process.env.DATABASE_URL !== url) {
@@ -67,6 +72,16 @@ async function rowsVersion(tableId: string): Promise<number> {
   return Number(row.rows_version)
 }
 
+interface QueryPlan {
+  'Node Type': string
+  'Index Name'?: string
+  Plans?: QueryPlan[]
+}
+
+function planNodes(plan: QueryPlan): QueryPlan[] {
+  return [plan, ...(plan.Plans ?? []).flatMap(planNodes)]
+}
+
 const textColumns = (...ids: string[]): ColumnDefinition[] =>
   ids.map((id) => ({ id, name: id, type: 'string' }))
 
@@ -342,6 +357,66 @@ describe.skipIf(!migrated)('table row writes against real PostgreSQL', () => {
         WHERE table_id = ${table.id} AND id IN ${control(expected)}`
       expect(remaining).toHaveLength(0)
     })
+
+    it('updates exactly the first matching rows for a filter the GIN index cannot serve', async () => {
+      const table = await createTable(textColumns('kind', 'n', 'flag'))
+      await seedInterleaved(table)
+      const expected = await firstHits(table.id, 9)
+      expect(expected).toHaveLength(9)
+
+      const result = await updateRowsByFilter(
+        table,
+        {
+          filter: { kind: { $ne: 'miss' } },
+          data: { flag: 'set' },
+          limit: 9,
+          secretProvenance: undefined,
+          capabilityGovernedUserId: null,
+        },
+        'limited-update-ne'
+      )
+
+      expect([...result.affectedRowIds].sort()).toEqual([...expected].sort())
+    })
+
+    it('fences only a containment filter into a GIN probe and leaves any other filter unfenced', async () => {
+      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},
+          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`
+
+      async function planFor(filter: Filter): Promise<QueryPlan[]> {
+        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)}`
+          )
+        )
+        return planNodes(plans[0]['QUERY PLAN'][0].Plan)
+      }
+
+      const fenced = await planFor({ kind: 'hit' })
+      expect(fenced.some((node) => node['Node Type'] === 'CTE Scan')).toBe(true)
+      expect(
+        fenced.some(
+          (node) =>
+            node['Node Type'] === 'Bitmap Index Scan' &&
+            node['Index Name'] === 'user_table_rows_tenant_data_gin_idx'
+        )
+      ).toBe(true)
+
+      const walked = await planFor({ kind: { $ne: 'miss' } })
+      expect(walked.some((node) => node['Node Type'] === 'CTE Scan')).toBe(false)
+      expect(walked[0]['Node Type']).toBe('Limit')
+    })
   })
 
   describe('unique columns under concurrent inserts', () => {

apps/sim/lib/table/rows/service.ts

@@ -73,6 +73,7 @@ import {
   buildSortClause,
   escapeLikePattern,
   fieldPredicate,
+  isContainmentOnlyFilter,
 } from '@/lib/table/sql'
 import { fireTableTrigger } from '@/lib/table/trigger'
 import { scaledStatementTimeoutMs, setTableTxTimeouts } from '@/lib/table/tx'
@@ -981,35 +982,59 @@ function buildRowOrderBySql(
 }
 
 /**
- * The first `limit` rows matching `filterClause` in the default `(order_key, id)` order.
+ * The first `limit` rows matching `filter` in the default `(order_key, id)` order.
  *
- * As one ordered query, the planner walks the `(table_id, order_key, id)` index and tests every
- * row against the filter until `limit` match — the whole table when matches are sparse. The
- * MATERIALIZED CTE fences the match into its own tenant-GIN probe; only the matches' sort keys
- * are sorted, and only the kept rows' `data` is read.
+ * As one ordered query, the planner walks the `(table_id, order_key, id)` index and tests each row
+ * against the filter until `limit` match: cheap when matches are dense, the whole table when they
+ * are sparse. A containment-only filter is fenced into a MATERIALIZED CTE instead, so the match
+ * runs as its own tenant-GIN probe and only the matches' sort keys are sorted. Any other filter
+ * cannot use the GIN index, so fencing it would evaluate and sort every match before the limit;
+ * it keeps the ordered walk, which stops early.
  */
+export function firstMatchingRowsQuery(
+  table: TableDefinition,
+  filter: Filter,
+  filterClause: SQL,
+  limit: number
+): SQL {
+  const tenantMatch = sql`${userTableRows.tableId} = ${table.id}
+    AND ${userTableRows.workspaceId} = ${table.workspaceId}
+    AND ${filterClause}`
+  if (!isContainmentOnlyFilter(filter, table.schema.columns)) {
+    return sql`
+      SELECT ${userTableRows.id}, ${userTableRows.data}
+      FROM ${userTableRows}
+      WHERE ${tenantMatch}
+      ORDER BY ${userTableRows.orderKey}, ${userTableRows.id}
+      LIMIT ${limit}
+    `
+  }
+  return sql`
+    WITH matched AS MATERIALIZED (
+      SELECT ${userTableRows.id}, ${userTableRows.orderKey}
+      FROM ${userTableRows}
+      WHERE ${tenantMatch}
+    ),
+    kept AS (
+      SELECT id, order_key FROM matched ORDER BY order_key, id LIMIT ${limit}
+    )
+    SELECT ${userTableRows.id}, ${userTableRows.data}
+    FROM kept
+    INNER JOIN ${userTableRows} ON ${userTableRows.id} = kept.id
+    ORDER BY kept.order_key, kept.id
+  `
+}
+
 async function selectFirstMatchingRows(
   table: TableDefinition,
+  filter: Filter,
   filterClause: SQL,
   limit: number
 ): Promise<Array<{ id: string; data: RowData }>> {
   const rows = await withSeqscanOff(async (trx) =>
-    trx.execute<{ id: string; data: RowData }>(sql`
-      WITH matched AS MATERIALIZED (
-        SELECT ${userTableRows.id}, ${userTableRows.orderKey}
-        FROM ${userTableRows}
-        WHERE ${userTableRows.tableId} = ${table.id}
-          AND ${userTableRows.workspaceId} = ${table.workspaceId}
-          AND ${filterClause}
-      ),
-      kept AS (
-        SELECT id, order_key FROM matched ORDER BY order_key, id LIMIT ${limit}
-      )
-      SELECT ${userTableRows.id}, ${userTableRows.data}
-      FROM kept
-      INNER JOIN ${userTableRows} ON ${userTableRows.id} = kept.id
-      ORDER BY kept.order_key, kept.id
-    `)
+    trx.execute<{ id: string; data: RowData }>(
+      firstMatchingRowsQuery(table, filter, filterClause, limit)
+    )
   )
   return Array.from(rows)
 }
@@ -2345,7 +2370,7 @@ export async function updateRowsByFilter(
     return { affectedCount: affectedRowIds.length, affectedRowIds }
   }
 
-  const matchingRows = await selectFirstMatchingRows(table, filterClause, limit)
+  const matchingRows = await selectFirstMatchingRows(table, data.filter, filterClause, limit)
   if (matchingRows.length === 0) {
     return { affectedCount: 0, affectedRowIds: [] }
   }
@@ -2720,7 +2745,7 @@ export async function deleteRowsByFilter(
       if (page.length < TABLE_LIMITS.DELETE_PAGE_SIZE) break
     }
   } else {
-    const matchingRows = await selectFirstMatchingRows(table, filterClause, limit)
+    const matchingRows = await selectFirstMatchingRows(table, data.filter, filterClause, limit)
     const rowIds = matchingRows.map((row) => row.id)
     if (rowIds.length > 0) {
       const deletedIds = await deleteOrderedRowsByIds({

apps/sim/lib/table/sql.ts

@@ -138,6 +138,38 @@ export function buildFilterClause(
   return buildFilterClauseInternal(filter, tableName, columnMap)
 }
 
+/**
+ * Whether every condition in `filter` compiles to JSONB containment, the only leaf the tenant
+ * `(table_id, data jsonb_path_ops)` GIN index serves. `$or`/`$and` groups qualify when all their
+ * members do. Mirrors the leaf choices in {@link fieldPredicate}: `eq`/`in` (and a multi-select's
+ * `contains`) compile to containment unless the column type compares through an equality cast;
+ * negations, ranges, patterns, emptiness checks, and system columns do not.
+ */
+export function isContainmentOnlyFilter(filter: Filter, columns: ColumnDefinition[]): boolean {
+  return filterIsContainmentOnly(filter, buildColumnMap(columns))
+}
+
+function filterIsContainmentOnly(filter: Filter, columnMap: ColumnMap): boolean {
+  return Object.entries(filter).every(([field, condition]) => {
+    if (condition === undefined) return true
+    if (field === '$or' || field === '$and') {
+      return (
+        Array.isArray(condition) &&
+        (condition as Filter[]).every((member) => filterIsContainmentOnly(member, columnMap))
+      )
+    }
+    if (Array.isArray(condition) || isSystemColumn(field)) return false
+    const column = columnMap.get(field)
+    const definition = column && columnTypeOf(column)
+    if (definition?.valueForEquality && definition.jsonbCast) return false
+    const isMultiSelect = column?.type === 'select' && column.multiple === true
+    const containmentOps = isMultiSelect ? ['$contains'] : ['$eq', '$in']
+    return isRecordLike(condition)
+      ? Object.keys(condition).every((op) => containmentOps.includes(op))
+      : true
+  })
+}
+
 function buildFilterClauseInternal(
   filter: Filter,
   tableName: string,

packages/db/migrations/0385_table_rows_version_at_commit.sql

@@ -13,7 +13,8 @@
 -- (cells) or `order_key` (row order). `id`, `table_id`, and `workspace_id` are
 -- never updated, and the snapshot never reads `position`. Constraint triggers
 -- are row-level only, so one transaction-local setting lists the tables already
--- bumped (a comma-delimited list of md5(table_id) tokens, matched exactly) and
+-- bumped (a comma-delimited list of table ids, matched exactly; ids are
+-- generated `tbl_<hex>` or UUIDs, so they never contain the delimiter) and
 -- dedupes the bump to one per table per transaction. It resets at transaction
 -- end, so each backend holds a single placeholder setting however many tables
 -- it writes. The row write and its bump still commit atomically, which the
@@ -26,12 +27,11 @@ CREATE OR REPLACE FUNCTION bump_user_table_rows_version_at_commit()
 RETURNS TRIGGER AS $$
 DECLARE
     bumped text := coalesce(current_setting('sim_rows_version.bumped', true), '');
-    token text := md5(NEW.table_id);
 BEGIN
-    IF position(',' || token || ',' IN ',' || bumped || ',') = 0 THEN
+    IF position(',' || NEW.table_id || ',' IN ',' || bumped || ',') = 0 THEN
         PERFORM set_config(
             'sim_rows_version.bumped',
-            CASE WHEN bumped = '' THEN token ELSE bumped || ',' || token END,
+            CASE WHEN bumped = '' THEN NEW.table_id ELSE bumped || ',' || NEW.table_id END,
             true
         );
         UPDATE user_table_definitions
@@ -64,16 +64,14 @@ CREATE CONSTRAINT TRIGGER user_table_rows_version_update_trigger
 -- determines workspace_id. Without a functional-dependency statistic the
 -- planner multiplies the two selectivities and underestimates a table's row
 -- count by orders of magnitude, which steers `data @>` filters to the btree
--- instead of the tenant GIN. A larger table_id sample keeps its most-common
--- values representative across many tables.
+-- instead of the tenant GIN. The per-column sample target stays at its
+-- default: the dependency alone restores the GIN plan, and a larger target
+-- would make every future autoanalyze of this table read more rows.
 
 CREATE STATISTICS IF NOT EXISTS user_table_rows_workspace_table_stats (dependencies)
     ON workspace_id, table_id FROM user_table_rows;
 --> statement-breakpoint
 
-ALTER TABLE user_table_rows ALTER COLUMN table_id SET STATISTICS 1000;
---> statement-breakpoint
-
 -- Release the DDL locks before the sample so writers never wait on it. The
 -- statements above are idempotent, so replaying this file after a partial run
 -- is safe.