Shared-scan graph projection: one scan per contract, declarative predicates (LLP 0095/0096)

projectGraph ran every contract rule as an independent full SQL scan with
refresh:'always', and the LLP 0026 aux filter prepended `attributes` to each
rule and JSON-parsed it per rule per row. At 25 rules that meant 25 full
table sweeps per projection: ~35GB read and 45+ minutes for a ~1.3GB source
on a production hypaware-server deployment (LLP 0095).

- project.js: one shared scan per contract over the union of declarative
  rules' columns; per-rule where predicates (eq/in/likePrefix, SQL null
  semantics) evaluated in JS; raw-SQL rules stay supported, run standalone,
  grouped by identical SQL text.
- Contract gains rowFilter, evaluated once per row on both paths; the
  ai-gateway contract's aux filter moves there.
- ai-gateway-graph: 21 of 25 rules migrated to columns/where; the two
  content_text prefix surfaces (x node+edge) stay raw SQL so the table's
  largest column keeps its server-side pushdown.
- Registry validates the new shapes; equivalence test pins the JS evaluator
  to the SQL engine's semantics over a fixture hitting every predicate.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: philcunliffe

hypaware-core/plugins-workspace/ai-gateway-graph/src/graph_contract.js

@@ -3,11 +3,24 @@
 /** A materialized graph row (node or edge), keyed by column name. */
 export type GraphRow = Record<string, unknown>
 
-/** One T0 contract rule: a read-only SELECT plus a row mapper. */
+/** A declarative rule filter (LLP 0096): AND of eq / in / likePrefix, SQL null semantics. */
+export interface RulePredicate {
+  eq?: Record<string, string>
+  in?: Record<string, string[]>
+  likePrefix?: Record<string, string>
+}
+
+/**
+ * One T0 contract rule: a source read plus a row mapper. Declarative
+ * `columns` (+ optional `where`) joins the contract's shared scan; raw `sql`
+ * runs standalone (LLP 0096). Exactly one of the two.
+ */
 export interface ContractRule {
   kind: 'node' | 'edge'
   type: string
-  sql: string
+  sql?: string
+  columns?: string[]
+  where?: RulePredicate
   toRow(row: Record<string, unknown>): GraphRow | null
 }
 

hypaware-core/plugins-workspace/ai-gateway-graph/src/types.d.ts

@@ -48,10 +48,24 @@ export function createContractRegistry(opts = {}) {
       throw new TypeError(`registerContract: '${contract.name}' rules must be a non-empty array`)
     }
     // Validate each rule's shape at registration, not at projection time: the
-    // engine reads `kind`/`sql`/`toRow` directly (project.js) and routes by
-    // `kind`, so a connector typo would otherwise surface as a confusing
-    // mid-projection failure (or silently route rows into the wrong target
-    // map) far from the contract that caused it.
+    // engine reads `kind`/`sql`/`columns`/`where`/`toRow` directly
+    // (project.js) and routes by `kind`, so a connector typo would otherwise
+    // surface as a confusing mid-projection failure (or silently route rows
+    // into the wrong target map) far from the contract that caused it.
+    // @ref LLP 0096#decision [implements]: exactly one read form per rule; `where` only rides `columns`; raw SQL must carry the rowFilter's columns itself
+    if (contract.rowFilter !== undefined) {
+      const filter = contract.rowFilter
+      const at = `'${contract.name}' rowFilter`
+      if (!filter || typeof filter !== 'object') {
+        throw new TypeError(`registerContract: ${at} must be an object`)
+      }
+      if (!Array.isArray(filter.columns) || filter.columns.length === 0 || filter.columns.some((c) => typeof c !== 'string' || c.length === 0)) {
+        throw new TypeError(`registerContract: ${at} columns must be non-empty strings`)
+      }
+      if (typeof filter.keep !== 'function') {
+        throw new TypeError(`registerContract: ${at} keep must be a function`)
+      }
+    }
     contract.rules.forEach((rule, i) => {
       const at = `'${contract.name}' rule ${i}`
       if (!rule || typeof rule !== 'object') {
@@ -63,8 +77,26 @@ export function createContractRegistry(opts = {}) {
       if (typeof rule.type !== 'string' || rule.type.length === 0) {
         throw new TypeError(`registerContract: ${at} type must be a non-empty string`)
       }
-      if (typeof rule.sql !== 'string' || rule.sql.length === 0) {
-        throw new TypeError(`registerContract: ${at} sql must be a non-empty string`)
+      const hasSql = typeof rule.sql === 'string' && rule.sql.length > 0
+      const hasColumns = Array.isArray(rule.columns)
+      if (hasSql === hasColumns) {
+        throw new TypeError(`registerContract: ${at} must carry exactly one of sql or columns`)
+      }
+      if (hasColumns) {
+        const cols = /** @type {unknown[]} */ (rule.columns)
+        if (cols.length === 0 || cols.some((c) => typeof c !== 'string' || c.length === 0)) {
+          throw new TypeError(`registerContract: ${at} columns must be non-empty strings`)
+        }
+        if (rule.where !== undefined) validatePredicate(rule.where, at)
+      } else if (rule.where !== undefined) {
+        throw new TypeError(`registerContract: ${at} where is only valid with columns`)
+      }
+      if (hasSql && contract.rowFilter) {
+        for (const col of contract.rowFilter.columns) {
+          if (!rule.sql?.includes(col)) {
+            throw new TypeError(`registerContract: ${at} raw sql must select rowFilter column '${col}'`)
+          }
+        }
       }
       if (typeof rule.toRow !== 'function') {
         throw new TypeError(`registerContract: ${at} toRow must be a function`)
@@ -86,6 +118,48 @@ export function createContractRegistry(opts = {}) {
     })
   }
 
+  /**
+   * A `where` must be built from the three supported predicate shapes only,
+   * with the value types the JS evaluator expects: anything else would
+   * silently match nothing at projection time.
+   *
+   * @param {unknown} where
+   * @param {string} at
+   */
+  function validatePredicate(where, at) {
+    if (!where || typeof where !== 'object') {
+      throw new TypeError(`registerContract: ${at} where must be an object`)
+    }
+    const w = /** @type {Record<string, unknown>} */ (where)
+    for (const key of Object.keys(w)) {
+      if (key !== 'eq' && key !== 'in' && key !== 'likePrefix') {
+        throw new TypeError(`registerContract: ${at} where.${key} is not a supported predicate (eq, in, likePrefix)`)
+      }
+    }
+    for (const shape of ['eq', 'likePrefix']) {
+      const block = w[shape]
+      if (block === undefined) continue
+      if (!block || typeof block !== 'object') {
+        throw new TypeError(`registerContract: ${at} where.${shape} must be an object`)
+      }
+      for (const [col, value] of Object.entries(block)) {
+        if (typeof value !== 'string' || value.length === 0) {
+          throw new TypeError(`registerContract: ${at} where.${shape}.${col} must be a non-empty string`)
+        }
+      }
+    }
+    if (w.in !== undefined) {
+      if (!w.in || typeof w.in !== 'object') {
+        throw new TypeError(`registerContract: ${at} where.in must be an object`)
+      }
+      for (const [col, list] of Object.entries(w.in)) {
+        if (!Array.isArray(list) || list.length === 0 || list.some((v) => typeof v !== 'string' || v.length === 0)) {
+          throw new TypeError(`registerContract: ${at} where.in.${col} must be a non-empty array of strings`)
+        }
+      }
+    }
+  }
+
   /**
    * All registered contracts, name-sorted so projection order is stable.
    * @returns {Contract[]}

hypaware-core/plugins-workspace/context-graph/src/contract-registry.js

@@ -45,33 +45,97 @@ export async function projectGraph({ query, storage, contracts, config, dryRun =
       /** @type {Map<string, GraphRow>} */
       const edges = new Map()
 
+      /**
+       * Feed one source row through a set of rules into the node/edge maps.
+       * Merge is order-independent (`mergeRow`), so fanning a shared scan's
+       * row out to many rules is observationally identical to the old
+       * one-query-per-rule order.
+       *
+       * @param {Record<string, unknown>} row
+       * @param {import('./types.js').ContractRule[]} rules
+       */
+      const applyRules = (row, rules) => {
+        for (const rule of rules) {
+          if (rule.where && !matchesPredicate(rule.where, row)) continue
+          const built = rule.toRow(row)
+          if (!built) continue
+          const target = rule.kind === 'node' ? nodes : edges
+          const idKey = rule.kind === 'node' ? 'node_id' : 'edge_id'
+          const id = /** @type {string} */ (built[idKey])
+          const existing = target.get(id)
+          if (existing) mergeRow(existing, built)
+          else target.set(id, built)
+        }
+      }
+
       let sourceRows = 0
+      let scanCount = 0
       for (const contract of contracts) {
-        for (const rule of contract.rules) {
+        const keep = contract.rowFilter?.keep
+        const declarative = contract.rules.filter((rule) => rule.columns)
+        const raw = contract.rules.filter((rule) => typeof rule.sql === 'string')
+
+        // One shared scan per contract: the union of the declarative rules'
+        // columns (plus the row filter's), each rule's predicate evaluated
+        // in JS per row. This is the whole LLP 0095 fix: rule count no
+        // longer multiplies table scans, and the contract row filter runs
+        // once per row instead of once per rule per row.
+        // @ref LLP 0096#decision [implements]: one scan per contract; JS predicates with SQL null semantics; rowFilter once per row
+        if (declarative.length > 0) {
+          const columns = new Set()
+          for (const rule of declarative) {
+            for (const col of rule.columns ?? []) columns.add(col)
+            for (const col of predicateColumns(rule.where)) columns.add(col)
+          }
+          for (const col of contract.rowFilter?.columns ?? []) columns.add(col)
           const result = await executeQuerySql({
-            query: rule.sql,
+            query: `SELECT ${[...columns].sort().join(', ')} FROM ${contract.sourceDataset}`,
             registry: query,
             storage,
             config,
             refresh: 'always',
           })
           sourceRows += result.rows.length
-          const target = rule.kind === 'node' ? nodes : edges
-          const idKey = rule.kind === 'node' ? 'node_id' : 'edge_id'
+          scanCount += 1
+          for (const row of result.rows) {
+            if (keep && !keep(row)) continue
+            applyRules(row, declarative)
+          }
+        }
+
+        // Raw-SQL rules run standalone, grouped by identical SQL text so
+        // rule pairs sharing a query (a surface's node + edge rule) cost one
+        // scan. Their SQL already selects the rowFilter's columns (registry
+        // enforced), so the filter applies here too.
+        /** @type {Map<string, import('./types.js').ContractRule[]>} */
+        const bySql = new Map()
+        for (const rule of raw) {
+          const sql = /** @type {string} */ (rule.sql)
+          const group = bySql.get(sql)
+          if (group) group.push(rule)
+          else bySql.set(sql, [rule])
+        }
+        for (const [sql, rules] of bySql) {
+          const result = await executeQuerySql({
+            query: sql,
+            registry: query,
+            storage,
+            config,
+            refresh: 'always',
+          })
+          sourceRows += result.rows.length
+          scanCount += 1
           for (const row of result.rows) {
-            const built = rule.toRow(row)
-            if (!built) continue
-            const id = /** @type {string} */ (built[idKey])
-            const existing = target.get(id)
-            if (existing) mergeRow(existing, built)
-            else target.set(id, built)
+            if (keep && !keep(row)) continue
+            applyRules(row, rules)
           }
         }
       }
 
       const nodeRows = [...nodes.values()]
       const edgeRows = [...edges.values()]
       span.setAttribute('contract_count', contracts.length)
+      span.setAttribute('scan_count', scanCount)
       span.setAttribute('source_row_count', sourceRows)
       span.setAttribute('node_count', nodeRows.length)
       span.setAttribute('edge_count', edgeRows.length)
@@ -144,6 +208,54 @@ async function dedupExisting(rows, idCol, dataset, query, storage, config) {
   return rows.filter((r) => !seen.has(/** @type {string} */ (r[idCol])))
 }
 
+/**
+ * Evaluate a rule's declarative predicate against one source row with SQL
+ * semantics: every clause must hold, and a null or absent column never
+ * matches (`WHERE col = 'x'` skips null rows; so does this).
+ *
+ * @param {import('./types.js').RulePredicate} where
+ * @param {Record<string, unknown>} row
+ * @returns {boolean}
+ * @ref LLP 0096#decision [implements]: eq/in/likePrefix only, AND-composed, null never matches
+ */
+function matchesPredicate(where, row) {
+  if (where.eq) {
+    for (const [col, value] of Object.entries(where.eq)) {
+      if (row[col] !== value) return false
+    }
+  }
+  if (where.in) {
+    for (const [col, values] of Object.entries(where.in)) {
+      const v = row[col]
+      if (typeof v !== 'string' || !values.includes(v)) return false
+    }
+  }
+  if (where.likePrefix) {
+    for (const [col, prefix] of Object.entries(where.likePrefix)) {
+      const v = row[col]
+      if (typeof v !== 'string' || !v.startsWith(prefix)) return false
+    }
+  }
+  return true
+}
+
+/**
+ * The columns a predicate reads, so the shared scan selects them even when
+ * no rule's `columns` lists them (a rule may filter on a column `toRow`
+ * never touches).
+ *
+ * @param {import('./types.js').RulePredicate | undefined} where
+ * @returns {string[]}
+ */
+function predicateColumns(where) {
+  if (!where) return []
+  return [
+    ...Object.keys(where.eq ?? {}),
+    ...Object.keys(where.in ?? {}),
+    ...Object.keys(where.likePrefix ?? {}),
+  ]
+}
+
 /**
  * True when a dedup query failed because the dataset is not queryable
  * yet (unregistered, or its backing path is absent) rather than because

hypaware-core/plugins-workspace/context-graph/src/project.js

@@ -16,11 +16,37 @@ export interface SkippedPartition {
   reason: 'unreadable-cursor' | 'unexpected-layout' | 'concurrent-write'
 }
 
-/** One T0 contract rule: a read-only SELECT plus a row mapper. */
+/**
+ * A declarative rule filter: the conjunction of the three predicate shapes
+ * contract rules actually use, evaluated in JS against the contract's one
+ * shared scan with SQL semantics (a null or absent column never matches).
+ * No general SQL parsing, no expression language (LLP 0096 §1).
+ */
+export interface RulePredicate {
+  /** column = value */
+  eq?: Record<string, string>
+  /** column IN (values) */
+  in?: Record<string, string[]>
+  /** column LIKE 'prefix%' */
+  likePrefix?: Record<string, string>
+}
+
+/**
+ * One T0 contract rule: a source read plus a row mapper. Exactly one read
+ * form: declarative `columns` (+ optional `where`) joins the contract's one
+ * shared scan; raw `sql` runs standalone, for contracts that migrate on
+ * their own schedule or predicates that must prune a heavy column
+ * server-side (LLP 0096 §2-3).
+ */
 export interface ContractRule {
   kind: 'node' | 'edge'
   type: string
-  sql: string
+  /** Raw SELECT escape hatch; mutually exclusive with `columns`. Must select the contract rowFilter's columns itself. */
+  sql?: string
+  /** Columns `toRow` reads; the shared scan selects the union across rules. Mutually exclusive with `sql`. */
+  columns?: string[]
+  /** Declarative filter; only valid with `columns`. */
+  where?: RulePredicate
   toRow(row: Record<string, unknown>): GraphRow | null
 }
 
@@ -82,6 +108,16 @@ export interface Contract {
   projectorVersion: number
   /** The node/edge rules the engine runs for this source. */
   rules: ContractRule[]
+  /**
+   * Optional source-row filter evaluated once per row before any rule's
+   * `toRow`, on both the shared-scan and raw-SQL paths (LLP 0096 §4). The
+   * engine adds `columns` to the shared scan; raw-SQL rules must select
+   * them in their own SQL (the registry enforces this textually).
+   */
+  rowFilter?: {
+    columns: string[]
+    keep(row: Record<string, unknown>): boolean
+  }
 }
 
 /** The in-plugin registry source plugins contribute contracts into. */