Repository: ponder-sh/ponder
Pull request #1596 (2025-03-11 18:36:58)

indexing store prediction
Improve the indexing store performance by recording historical cache accesses and using the data to prediction future accesses.


Patch:
diff --git a/.changeset/light-owls-judge.md b/.changeset/light-owls-judge.md
new file mode 100644
index 00000000..aede5d1e
--- /dev/null
+++ b/.changeset/light-owls-judge.md
@@ -0,0 +1,5 @@
+---
+"ponder": patch
+---
+
+Improved historical indexing performance.
diff --git a/packages/client/CHANGELOG.md b/packages/client/CHANGELOG.md
index b4e4b7e6..05f3ac04 100644
--- a/packages/client/CHANGELOG.md
+++ b/packages/client/CHANGELOG.md
@@ -1,5 +1,7 @@
 # @ponder/client
 
+## 0.10.7
+
 ## 0.10.6
 
 ## 0.10.5
diff --git a/packages/client/package.json b/packages/client/package.json
index 6a25f4de..2836053c 100644
--- a/packages/client/package.json
+++ b/packages/client/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@ponder/client",
-  "version": "0.10.6",
+  "version": "0.10.7",
   "description": "",
   "license": "MIT",
   "type": "module",
diff --git a/packages/core/CHANGELOG.md b/packages/core/CHANGELOG.md
index 57f8f71d..40b75ac7 100644
--- a/packages/core/CHANGELOG.md
+++ b/packages/core/CHANGELOG.md
@@ -1,5 +1,13 @@
 # ponder
 
+## 0.10.7
+
+### Patch Changes
+
+- [#1631](https://github.com/ponder-sh/ponder/pull/1631) [`4045aea473eb68b304357851fa47c9a165a00e49`](https://github.com/ponder-sh/ponder/commit/4045aea473eb68b304357851fa47c9a165a00e49) Thanks [@kyscott18](https://github.com/kyscott18)! - Improved raw SQL performance.
+
+- [#1634](https://github.com/ponder-sh/ponder/pull/1634) [`faac60dbb9b201639a91e3b5ddcd50ea83c21860`](https://github.com/ponder-sh/ponder/commit/faac60dbb9b201639a91e3b5ddcd50ea83c21860) Thanks [@kyscott18](https://github.com/kyscott18)! - Improved performance for factories during the historical sync.
+
 ## 0.10.6
 
 ### Patch Changes
diff --git a/packages/core/package.json b/packages/core/package.json
index 45172608..f2afba09 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -1,6 +1,6 @@
 {
   "name": "ponder",
-  "version": "0.10.6",
+  "version": "0.10.7",
   "description": "An open-source framework for crypto application backends",
   "license": "MIT",
   "type": "module",
diff --git a/packages/core/src/bin/utils/run.ts b/packages/core/src/bin/utils/run.ts
index fae4b994..6ce4d84b 100644
--- a/packages/core/src/bin/utils/run.ts
+++ b/packages/core/src/bin/utils/run.ts
@@ -86,7 +86,7 @@ export async function run({
   const indexingCache = createIndexingCache({
     common,
     schemaBuild,
-    checkpoint: crashRecoveryCheckpoint,
+    crashRecoveryCheckpoint,
   });
 
   await database.setStatus(sync.getStatus());
@@ -161,8 +161,18 @@ export async function run({
       );
     },
   )) {
+    let endClock = startClock();
+    await indexingCache.prefetch({
+      events,
+      db: database.qb.drizzle,
+      eventCount: indexing.getEventCount(),
+    });
+    common.metrics.ponder_historical_transform_duration.inc(
+      { step: "prefetch" },
+      endClock(),
+    );
     if (events.length > 0) {
-      let endClock = startClock();
+      endClock = startClock();
       await database.retry(async () => {
         await database
           .transaction(async (client, tx) => {
@@ -185,6 +195,7 @@ export async function run({
               const result = await indexing.processEvents({
                 events: eventChunk,
                 db: historicalIndexingStore,
+                cache: indexingCache,
               });
 
               if (result.status === "error") {
@@ -240,6 +251,8 @@ export async function run({
             );
 
             endClock = startClock();
+            // Note: at this point, the next events can be preloaded, as long as the are not indexed until
+            // the "flush" + "finalize" is complete.
 
             try {
               await indexingCache.flush({ client });
@@ -273,7 +286,6 @@ export async function run({
             throw error;
           });
       });
-      indexingCache.commit();
       common.metrics.ponder_historical_transform_duration.inc(
         { step: "commit" },
         endClock(),
diff --git a/packages/core/src/client/validate.ts b/packages/core/src/client/validate.ts
index 72f9be2c..8602ec08 100644
--- a/packages/core/src/client/validate.ts
+++ b/packages/core/src/client/validate.ts
@@ -12,7 +12,16 @@ const getNodeType = (node: Node) => Object.keys(node)[0]!;
 
 const ALLOW_CACHE = new Map<string, boolean>();
 
-export const validateQuery = async (sql: string) => {
+/**
+ * Validate a SQL query.
+ *
+ * @param sql - SQL query
+ * @param shouldValidateInnerNode - `true` if the properties of each ast node should be validated, else only the allow list is checked
+ */
+export const validateQuery = async (
+  sql: string,
+  shouldValidateInnerNode = true,
+) => {
   // @ts-ignore
   const Parser = await import(/* webpackIgnore: true */ "pg-query-emscripten");
   const crypto = await import(/* webpackIgnore: true */ "node:crypto");
@@ -64,8 +73,10 @@ export const validateQuery = async (sql: string) => {
       throw new Error(`${getNodeType(node)} not supported`);
     }
 
-    // @ts-ignore
-    ALLOW_LIST.get(getNodeType(node))!.validate?.(node[getNodeType(node)]);
+    if (shouldValidateInnerNode) {
+      // @ts-ignore
+      ALLOW_LIST.get(getNodeType(node))!.validate?.(node[getNodeType(node)]);
+    }
 
     for (const child of ALLOW_LIST.get(getNodeType(node))!.children(
       // @ts-ignore
diff --git a/packages/core/src/database/index.ts b/packages/core/src/database/index.ts
index 9cc5b480..86d17fad 100644
--- a/packages/core/src/database/index.ts
+++ b/packages/core/src/database/index.ts
@@ -598,9 +598,7 @@ export const createDatabase = async ({
     },
     async transaction(fn) {
       if (dialect === "postgres") {
-        const client = await (
-          database.driver as { internal: Pool }
-        ).internal.connect();
+        const client = await (database.driver as { user: Pool }).user.connect();
         try {
           await client.query("BEGIN");
           const tx = drizzleNodePg(client, {
diff --git a/packages/core/src/indexing-store/cache.ts b/packages/core/src/indexing-store/cache.ts
index 88e2c6eb..58ef2869 100644
--- a/packages/core/src/indexing-store/cache.ts
+++ b/packages/core/src/indexing-store/cache.ts
@@ -8,6 +8,7 @@ import { FlushError } from "@/internal/errors.js";
 import type { Event, Schema, SchemaBuild } from "@/internal/types.js";
 import type { Drizzle } from "@/types/db.js";
 import { ZERO_CHECKPOINT_STRING } from "@/utils/checkpoint.js";
+import { dedupe } from "@/utils/dedupe.js";
 import { prettyPrint } from "@/utils/print.js";
 import { startClock } from "@/utils/timer.js";
 import { PGlite } from "@electric-sql/pglite";
@@ -18,6 +19,7 @@ import {
   getTableColumns,
   getTableName,
   is,
+  or,
 } from "drizzle-orm";
 import {
   PgTable,
@@ -27,6 +29,11 @@ import {
 import type { PoolClient } from "pg";
 import copy from "pg-copy-streams";
 import { parseSqlError } from "./index.js";
+import {
+  getProfilePatternKey,
+  recordProfilePattern,
+  recoverProfilePattern,
+} from "./profile.js";
 import { getCacheKey, getWhereCondition, normalizeRow } from "./utils.js";
 
 export type IndexingCache = {
@@ -41,22 +48,16 @@ export type IndexingCache = {
     table: Table;
     key: object;
     db: Drizzle<Schema>;
-  }) =>
-    | { [key: string]: unknown }
-    | null
-    | Promise<{ [key: string]: unknown } | null>;
+  }) => Row | null | Promise<Row | null>;
   /**
    * Sets the entry for `table` with `key` to `row`.
    */
   set: (params: {
     table: Table;
     key: object;
-    row: { [key: string]: unknown };
+    row: Row;
     isUpdate: boolean;
-    metadata: {
-      event: Event | undefined;
-    };
-  }) => { [key: string]: unknown };
+  }) => Row;
   /**
    * Deletes the entry for `table` with `key`.
    */
@@ -70,13 +71,13 @@ export type IndexingCache = {
    */
   flush: (params: { client: PoolClient | PGlite }) => Promise<void>;
   /**
-   * Make all temporary data permanent and prepare the cache for
-   * the next iteration.
-   *
-   * Note: It is assumed this is called after `flush`
-   * because it clears the buffers.
+   * Predict and load rows that will be accessed in the next event batch.
    */
-  commit: () => void;
+  prefetch: (params: {
+    events: Event[];
+    db: Drizzle<Schema>;
+    eventCount: { [key: string]: number };
+  }) => Promise<void>;
   /**
    * Remove spillover and buffer entries.
    */
@@ -89,36 +90,81 @@ export type IndexingCache = {
    * Deletes all entries from the cache.
    */
   clear: () => void;
+  event: Event | undefined;
 };
 
-type Cache = Map<
-  Table,
-  Map<
-    string,
-    {
-      bytes: number;
-      operationIndex: number;
-      row: { [key: string]: unknown } | null;
-    }
-  >
->;
-
+/**
+ * Database row.
+ *
+ * @example
+ * {
+ *   "owner": "0x123",
+ *   "spender": "0x456",
+ *   "amount": 100n,
+ * }
+ */
+export type Row = { [key: string]: unknown };
+/**
+ * Serialized primary key values for uniquely identifying a database row.
+ *
+ * @example
+ * "0x123_0x456"
+ */
+type CacheKey = string;
+/**
+ * Event name.
+ *
+ * @example
+ * "Erc20:Transfer"
+ *
+ * @example
+ * "Erc20.mint()"
+ */
+type EventName = string;
+/**
+ * Recorded database access pattern.
+ *
+ * @example
+ * {
+ *   "owner": "args.from",
+ *   "spender": "log.address",
+ * }
+ */
+export type ProfilePattern = { [key: string]: string };
+/**
+ * Serialized for uniquely identifying a {@link ProfilePattern}.
+ *
+ * @example
+ * "args.from_spender_log.address"
+ */
+type ProfileKey = string;
+/**
+ * Cache of database rows.
+ */
+type Cache = Map<Table, Map<CacheKey, Row | null>>;
+/**
+ * Buffer of database rows that will be flushed to the database.
+ */
 type Buffer = Map<
   Table,
   Map<
-    string,
+    CacheKey,
     {
-      row: { [key: string]: unknown };
-      metadata: {
-        event: Event | undefined;
-      };
+      row: Row;
+      metadata: { event: Event | undefined };
     }
   >
 >;
+/**
+ * Metadata about database access patterns for each event.
+ */
+type Profile = Map<
+  EventName,
+  Map<Table, Map<ProfileKey, { pattern: ProfilePattern; count: number }>>
+>;
 
 const getBytes = (value: unknown) => {
-  // size of metadata
-  let size = 13;
+  let size = 0;
 
   if (typeof value === "number") {
     size += 8;
@@ -145,10 +191,7 @@ const getBytes = (value: unknown) => {
 
 const ESCAPE_REGEX = /([\\\b\f\n\r\t\v])/g;
 
-export const getCopyText = (
-  table: Table,
-  rows: { [key: string]: unknown }[],
-) => {
+export const getCopyText = (table: Table, rows: Row[]) => {
   const columns = Object.entries(getTableColumns(table));
   const results = new Array(rows.length);
   for (let i = 0; i < rows.length; i++) {
@@ -232,39 +275,37 @@ export const recoverBatchError = async <T>(
 export const createIndexingCache = ({
   common,
   schemaBuild: { schema },
-  checkpoint,
+  crashRecoveryCheckpoint,
 }: {
   common: Common;
   schemaBuild: Pick<SchemaBuild, "schema">;
-  checkpoint: string;
+  crashRecoveryCheckpoint: string;
 }): IndexingCache => {
+  /**
+   * Estimated size of the cache in bytes.
+   *
+   * Note: this stops getting updated once `isCacheComplete = false`.
+   */
+  let cacheBytes = 0;
+  let event: Event | undefined;
+  let isCacheComplete = crashRecoveryCheckpoint === ZERO_CHECKPOINT_STRING;
+  const primaryKeyCache = new Map<Table, [string, Column][]>();
+
   const cache: Cache = new Map();
-  const spillover: Cache = new Map();
   const insertBuffer: Buffer = new Map();
   const updateBuffer: Buffer = new Map();
+  /** Metadata about which entries in cache were not prefetched but were accessed anyway. */
+  const spillover: Map<Table, Set<string>> = new Map();
+  /** Profiling data about access patterns for each event. */
+  const profile: Profile = new Map();
 
-  const primaryKeyCache = new Map<Table, [string, Column][]>();
-
-  let isCacheComplete = checkpoint === ZERO_CHECKPOINT_STRING;
-
-  let cacheBytes = 0;
-  let spilloverBytes = 0;
-
-  // LRU counter
-  let totalCacheOps = 0;
-
-  common.logger.debug({
-    service: "indexing",
-    msg: `Using a ${Math.round(
-      common.options.indexingCacheMaxBytes / (1024 * 1024),
-    )} MB indexing cache`,
-  });
-
-  for (const table of Object.values(schema).filter(
+  const tables = Object.values(schema).filter(
     (table): table is PgTableWithColumns<TableConfig> => is(table, PgTable),
-  )) {
+  );
+
+  for (const table of tables) {
     cache.set(table, new Map());
-    spillover.set(table, new Map());
+    spillover.set(table, new Set());
     insertBuffer.set(table, new Map());
     updateBuffer.set(table, new Map());
 
@@ -281,12 +322,41 @@ export const createIndexingCache = ({
 
       return (
         cache.get(table)!.has(ck) ??
-        spillover.get(table)!.has(ck) ??
         insertBuffer.get(table)!.has(ck) ??
         updateBuffer.get(table)!.has(ck)
       );
     },
     async get({ table, key, db }) {
+      if (event) {
+        if (profile.has(event.name) === false) {
+          profile.set(event.name, new Map());
+          for (const table of tables) {
+            profile.get(event.name)!.set(table, new Map());
+          }
+        }
+
+        const pattern = recordProfilePattern(
+          event,
+          table,
+          key,
+          Array.from(profile.get(event.name)!.get(table)!.values()).map(
+            ({ pattern }) => pattern,
+          ),
+          primaryKeyCache,
+        );
+        if (pattern) {
+          const key = getProfilePatternKey(pattern);
+          if (profile.get(event.name)!.get(table)!.has(key)) {
+            profile.get(event.name)!.get(table)!.get(key)!.count++;
+          } else {
+            profile
+              .get(event.name)!
+              .get(table)!
+              .set(key, { pattern, count: 1 });
+          }
+        }
+      }
+
       const ck = getCacheKey(table, key, primaryKeyCache);
       // Note: order is important, it is an invariant that update entries
       // are prioritized over insert entries
@@ -296,33 +366,31 @@ export const createIndexingCache = ({
       if (bufferEntry) {
         common.metrics.ponder_indexing_cache_requests_total.inc({
           table: getTableName(table),
-          type: "hit",
+          type: isCacheComplete ? "complete" : "hit",
         });
         return structuredClone(bufferEntry.row);
       }
 
-      const entry = spillover.get(table)!.get(ck) ?? cache.get(table)!.get(ck);
-
-      if (entry) {
-        entry.operationIndex = totalCacheOps++;
+      const entry = cache.get(table)!.get(ck);
 
-        if (entry.row) {
-          common.metrics.ponder_indexing_cache_requests_total.inc({
-            table: getTableName(table),
-            type: "hit",
-          });
-          return structuredClone(entry.row);
-        }
+      if (entry !== undefined) {
+        common.metrics.ponder_indexing_cache_requests_total.inc({
+          table: getTableName(table),
+          type: isCacheComplete ? "complete" : "hit",
+        });
+        return structuredClone(entry);
       }
 
       if (isCacheComplete) {
         common.metrics.ponder_indexing_cache_requests_total.inc({
           table: getTableName(table),
-          type: "hit",
+          type: "complete",
         });
         return null;
       }
 
+      spillover.get(table)!.add(ck);
+
       common.metrics.ponder_indexing_cache_requests_total.inc({
         table: getTableName(table),
         type: "miss",
@@ -336,14 +404,11 @@ export const createIndexingCache = ({
         .where(getWhereCondition(table, key))
         .then((res) => (res.length === 0 ? null : res[0]!))
         .then((row) => {
-          const bytes = getBytes(row);
-          spilloverBytes += bytes;
+          cache.get(table)!.set(ck, structuredClone(row));
+
+          // Note: the size is not recorded because it is not possible
+          // to miss the cache when in the "full in-memory" mode
 
-          spillover.get(table)!.set(ck, {
-            bytes,
-            operationIndex: totalCacheOps++,
-            row: structuredClone(row),
-          });
           return row;
         });
 
@@ -354,22 +419,21 @@ export const createIndexingCache = ({
         },
         endClock(),
       );
-
       return result;
     },
-    set({ table, key, row: _row, isUpdate, metadata }) {
+    set({ table, key, row: _row, isUpdate }) {
       const row = normalizeRow(table, _row, isUpdate);
       const ck = getCacheKey(table, key, primaryKeyCache);
 
       if (isUpdate) {
         updateBuffer.get(table)!.set(ck, {
           row: structuredClone(row),
-          metadata,
+          metadata: { event },
         });
       } else {
         insertBuffer.get(table)!.set(ck, {
           row: structuredClone(row),
-          metadata,
+          metadata: { event },
         });
       }
 
@@ -382,7 +446,6 @@ export const createIndexingCache = ({
       const inUpdateBuffer = updateBuffer.get(table)!.delete(ck);
 
       cache.get(table)!.delete(ck);
-      spillover.get(table)!.delete(ck);
 
       const inDb = await db
         .delete(table)
@@ -395,8 +458,10 @@ export const createIndexingCache = ({
     async flush({ client }) {
       const copy = getCopyHelper({ client });
 
+      const shouldRecordBytes = isCacheComplete;
+
       for (const table of cache.keys()) {
-        const tableSpillover = spillover.get(table)!;
+        const tableCache = cache.get(table)!;
 
         const insertValues = Array.from(insertBuffer.get(table)!.values());
         const updateValues = Array.from(updateBuffer.get(table)!.values());
@@ -470,14 +535,10 @@ export const createIndexingCache = ({
           }
 
           for (const [key, entry] of insertBuffer.get(table)!) {
-            const bytes = getBytes(entry.row);
-            cacheBytes += bytes;
-            totalCacheOps++;
-            tableSpillover.set(key, {
-              bytes,
-              operationIndex: 0,
-              row: entry.row,
-            });
+            if (shouldRecordBytes && tableCache.has(key) === false) {
+              cacheBytes += getBytes(entry.row);
+            }
+            tableCache.set(key, entry.row);
           }
           insertBuffer.get(table)!.clear();
 
@@ -602,13 +663,10 @@ export const createIndexingCache = ({
           );
 
           for (const [key, entry] of updateBuffer.get(table)!) {
-            const bytes = getBytes(entry.row);
-            cacheBytes += bytes;
-            tableSpillover.set(key, {
-              bytes,
-              operationIndex: totalCacheOps++,
-              row: entry.row,
-            });
+            if (shouldRecordBytes && tableCache.has(key) === false) {
+              cacheBytes += getBytes(entry.row);
+            }
+            tableCache.set(key, entry.row);
           }
           updateBuffer.get(table)!.clear();
 
@@ -626,50 +684,130 @@ export const createIndexingCache = ({
         }
       }
     },
-    commit() {
-      for (const [table, tableSpillover] of spillover) {
-        const tableCache = cache.get(table)!;
-        for (const [key, entry] of tableSpillover) {
-          tableCache.set(key, entry);
+    async prefetch({ events, db, eventCount }) {
+      if (isCacheComplete) {
+        if (cacheBytes < common.options.indexingCacheMaxBytes) {
+          return;
         }
-        tableSpillover.clear();
-      }
-
-      cacheBytes += spilloverBytes;
-      spilloverBytes = 0;
 
-      let cacheSize = 0;
-      for (const { size } of cache.values()) cacheSize += size;
+        isCacheComplete = false;
+        for (const table of cache.keys()) {
+          cache.get(table)!.clear();
+          // Note: spillover is not cleared because it is an invariant
+          // it is empty
+        }
+      }
 
-      const flushIndex =
-        totalCacheOps -
-        cacheSize * (1 - common.options.indexingCacheEvictRatio);
+      // Use historical accesses + next event batch to determine which
+      // rows are going to be accessed, and preload them into the cache.
 
-      if (cacheBytes + spilloverBytes > common.options.indexingCacheMaxBytes) {
-        isCacheComplete = false;
+      const prediction = new Map<Table, Map<CacheKey, Row>>();
 
-        let rowCount = 0;
+      for (const table of tables) {
+        prediction.set(table, new Map());
+      }
 
-        for (const tableCache of cache.values()) {
-          for (const [key, entry] of tableCache) {
-            if (entry.operationIndex <= flushIndex) {
-              tableCache.delete(key);
-              cacheBytes -= entry.bytes;
-              rowCount += 1;
+      for (const event of events) {
+        if (profile.has(event.name)) {
+          for (const table of tables) {
+            for (const [, { count, pattern }] of profile
+              .get(event.name)!
+              .get(table)!) {
+              // Expected value of times the prediction will be used.
+              const ev = count / eventCount[event.name]!;
+              if (ev > 0.25) {
+                const row = recoverProfilePattern(pattern, event);
+                const key = getCacheKey(table, row);
+                prediction.get(table)!.set(key, row);
+              }
             }
           }
         }
+      }
 
-        common.logger.debug({
-          service: "indexing",
-          msg: `Evicted ${rowCount} rows from the cache`,
-        });
+      for (const [table, tableCache] of cache) {
+        for (const key of tableCache.keys()) {
+          if (
+            spillover.get(table)!.has(key) ||
+            prediction.get(table)!.has(key)
+          ) {
+            prediction.get(table)!.delete(key);
+          } else {
+            tableCache.delete(key);
+            isCacheComplete = false;
+          }
+        }
+      }
+
+      for (const [table] of spillover) {
+        spillover.get(table)!.clear();
+      }
+
+      for (const [table, tablePredictions] of prediction) {
+        common.metrics.ponder_indexing_cache_requests_total.inc(
+          {
+            table: getTableName(table),
+            type: "prefetch",
+          },
+          tablePredictions.size,
+        );
       }
+
+      await Promise.all(
+        Array.from(prediction.entries())
+          .filter(([, tablePredictions]) => tablePredictions.size > 0)
+          .map(async ([table, tablePredictions]) => {
+            const conditions = dedupe(
+              Array.from(tablePredictions),
+              ([key]) => key,
+            ).map(([, prediction]) => getWhereCondition(table, prediction));
+
+            if (conditions.length === 0) return;
+            const endClock = startClock();
+
+            await db
+              .select()
+              .from(table)
+              .where(or(...conditions))
+              .then((results) => {
+                common.logger.debug({
+                  service: "indexing",
+                  msg: `Pre-queried ${results.length} '${getTableName(table)}' rows`,
+                });
+                const resultsPerKey = new Map<CacheKey, Row>();
+                for (const result of results) {
+                  const ck = getCacheKey(table, result, primaryKeyCache);
+                  resultsPerKey.set(ck, result);
+                }
+
+                const tableCache = cache.get(table)!;
+                for (const key of tablePredictions.keys()) {
+                  if (resultsPerKey.has(key)) {
+                    tableCache.set(key, resultsPerKey.get(key)!);
+                  } else {
+                    tableCache.set(key, null);
+                  }
+                }
+              });
+
+            common.metrics.ponder_indexing_cache_query_duration.observe(
+              {
+                table: getTableName(table),
+                method: "load",
+              },
+              endClock(),
+            );
+          }),
+      );
     },
     invalidate() {
       isCacheComplete = false;
     },
     rollback() {
+      for (const tableCache of cache.values()) {
+        tableCache.clear();
+      }
+
       for (const tableSpillover of spillover.values()) {
         tableSpillover.clear();
       }
@@ -681,8 +819,6 @@ export const createIndexingCache = ({
       for (const tableBuffer of updateBuffer.values()) {
         tableBuffer.clear();
       }
-
-      spilloverBytes = 0;
     },
     clear() {
       for (const tableCache of cache.values()) {
@@ -700,9 +836,9 @@ export const createIndexingCache = ({
       for (const tableBuffer of updateBuffer.values()) {
         tableBuffer.clear();
       }
-
-      cacheBytes = 0;
-      spilloverBytes = 0;
+    },
+    set event(_event: Event | undefined) {
+      event = _event;
     },
   };
 };
diff --git a/packages/core/src/indexing-store/historical.ts b/packages/core/src/indexing-store/historical.ts
index 30a59d78..91c472dc 100644
--- a/packages/core/src/indexing-store/historical.ts
+++ b/packages/core/src/indexing-store/historical.ts
@@ -1,6 +1,7 @@
+import { validateQuery } from "@/client/validate.js";
 import type { Common } from "@/internal/common.js";
 import { RecordNotFoundError } from "@/internal/errors.js";
-import type { Event, Schema, SchemaBuild } from "@/internal/types.js";
+import type { Schema, SchemaBuild } from "@/internal/types.js";
 import type { Drizzle } from "@/types/db.js";
 import { prettyPrint } from "@/utils/print.js";
 import { startClock } from "@/utils/timer.js";
@@ -28,8 +29,6 @@ export const createHistoricalIndexingStore = ({
   db: Drizzle<Schema>;
   client: PoolClient | PGlite;
 }): IndexingStore => {
-  let event: Event | undefined;
-
   return {
     // @ts-ignore
     find: (table: Table, key) => {
@@ -72,7 +71,6 @@ export const createHistoricalIndexingStore = ({
                         key: value,
                         row: value,
                         isUpdate: false,
-                        metadata: { event },
                       }),
                     );
                   }
@@ -94,7 +92,6 @@ export const createHistoricalIndexingStore = ({
                   key: values,
                   row: values,
                   isUpdate: false,
-                  metadata: { event },
                 });
               }
             },
@@ -132,7 +129,6 @@ export const createHistoricalIndexingStore = ({
                         key: value,
                         row,
                         isUpdate: true,
-                        metadata: { event },
                       }),
                     );
                   } else {
@@ -142,7 +138,6 @@ export const createHistoricalIndexingStore = ({
                         key: value,
                         row: value,
                         isUpdate: false,
-                        metadata: { event },
                       }),
                     );
                   }
@@ -172,7 +167,6 @@ export const createHistoricalIndexingStore = ({
                     key: values,
                     row,
                     isUpdate: true,
-                    metadata: { event },
                   });
                 }
 
@@ -181,7 +175,6 @@ export const createHistoricalIndexingStore = ({
                   key: values,
                   row: values,
                   isUpdate: false,
-                  metadata: { event },
                 });
               }
             },
@@ -205,7 +198,6 @@ export const createHistoricalIndexingStore = ({
                       key: value,
                       row: value,
                       isUpdate: false,
-                      metadata: { event },
                     }),
                   );
                 }
@@ -219,7 +211,6 @@ export const createHistoricalIndexingStore = ({
                   key: values,
                   row: values,
                   isUpdate: false,
-                  metadata: { event },
                 });
                 return Promise.resolve(result).then(onFulfilled, onRejected);
               }
@@ -280,7 +271,6 @@ export const createHistoricalIndexingStore = ({
             key,
             row,
             isUpdate: true,
-            metadata: { event },
           });
         },
       };
@@ -298,8 +288,18 @@ export const createHistoricalIndexingStore = ({
     sql: drizzle(
       async (_sql, params, method, typings) => {
         await indexingCache.flush({ client });
-        indexingCache.invalidate();
-        indexingCache.clear();
+
+        let safeQuery = false;
+
+        try {
+          await validateQuery(_sql, false);
+          safeQuery = true;
+        } catch {}
+
+        if (safeQuery === false) {
+          indexingCache.invalidate();
+          indexingCache.clear();
+        }
 
         const query: QueryWithTypings = { sql: _sql, params, typings };
         const endClock = startClock();
@@ -321,8 +321,5 @@ export const createHistoricalIndexingStore = ({
       },
       { schema, casing: "snake_case" },
     ),
-    set event(_event: Event | undefined) {
-      event = _event;
-    },
   };
 };
diff --git a/packages/core/src/indexing-store/index.ts b/packages/core/src/indexing-store/index.ts
index c3fdebf5..19eb6819 100644
--- a/packages/core/src/indexing-store/index.ts
+++ b/packages/core/src/indexing-store/index.ts
@@ -8,14 +8,12 @@ import {
   UniqueConstraintError,
   getBaseError,
 } from "@/internal/errors.js";
-import type { Event, Schema } from "@/internal/types.js";
+import type { Schema } from "@/internal/types.js";
 import type { Db } from "@/types/db.js";
 import type { Table } from "drizzle-orm";
 import { getTableConfig } from "drizzle-orm/pg-core";
 
-export type IndexingStore = Db<Schema> & {
-  event: Event | undefined;
-};
+export type IndexingStore = Db<Schema>;
 
 export const parseSqlError = (e: any): Error => {
   let error = getBaseError(e);
diff --git a/packages/core/src/indexing-store/profile.ts b/packages/core/src/indexing-store/profile.ts
new file mode 100644
index 00000000..3798247d
--- /dev/null
+++ b/packages/core/src/indexing-store/profile.ts
@@ -0,0 +1,374 @@
+import type { Event } from "@/internal/types.js";
+import type { Column, Table } from "drizzle-orm";
+import type { ProfilePattern, Row } from "./cache.js";
+import { getCacheKey } from "./utils.js";
+
+export const getProfilePatternKey = (pattern: ProfilePattern): string => {
+  return Object.values(pattern).join("_");
+};
+
+const eq = (
+  target: bigint | string | number | boolean,
+  value: bigint | string | number | boolean | null | undefined,
+) => {
+  if (target === value) return true;
+  if (target && value && target.toString() === value.toString()) return true;
+  return false;
+};
+
+export const recordProfilePattern = (
+  event: Event,
+  table: Table,
+  key: object,
+  hints: { [key: string]: string }[],
+  cache: Map<Table, [string, Column][]>,
+): { [key: string]: string } | undefined => {
+  for (const hint of hints) {
+    if (
+      getCacheKey(table, key, cache) ===
+      getCacheKey(table, recoverProfilePattern(hint, event), cache)
+    ) {
+      return hint;
+    }
+  }
+
+  const result: { [key: string]: string } = {};
+
+  for (const [js] of cache.get(table)!) {
+    // @ts-ignore
+    const value = key[js]!;
+
+    if (eq(event.chainId, value)) {
+      result[js] = "chainId";
+      continue;
+    }
+
+    if (eq(event.event.id, value)) {
+      result[js] = "id";
+      continue;
+    }
+
+    switch (event.type) {
+      case "block": {
+        if (eq(event.event.block.hash, value)) {
+          result[js] = "block.hash";
+          continue;
+        }
+
+        if (eq(event.event.block.number, value)) {
+          result[js] = "block.number";
+          continue;
+        }
+
+        if (eq(event.event.block.timestamp, value)) {
+          result[js] = "block.timestamp";
+          continue;
+        }
+
+        if (eq(event.event.block.miner, value)) {
+          result[js] = "block.miner";
+          continue;
+        }
+
+        break;
+      }
+
+      case "transaction": {
+        if (eq(event.event.block.hash, value)) {
+          result[js] = "block.hash";
+          continue;
+        }
+
+        if (eq(event.event.block.number, value)) {
+          result[js] = "block.number";
+          continue;
+        }
+
+        if (eq(event.event.block.timestamp, value)) {
+          result[js] = "block.timestamp";
+          continue;
+        }
+
+        if (eq(event.event.block.miner, value)) {
+          result[js] = "block.miner";
+          continue;
+        }
+
+        if (eq(event.event.transaction.hash, value)) {
+          result[js] = "transaction.hash";
+          continue;
+        }
+
+        if (eq(event.event.transaction.from, value)) {
+          result[js] = "transaction.from";
+          continue;
+        }
+
+        if (
+          event.event.transaction.to &&
+          eq(event.event.transaction.to, value)
+        ) {
+          result[js] = "transaction.to";
+          continue;
+        }
+
+        if (eq(event.event.transaction.transactionIndex, value)) {
+          result[js] = "transaction.transactionIndex";
+          continue;
+        }
+
+        if (
+          event.event.transactionReceipt?.contractAddress &&
+          eq(event.event.transactionReceipt.contractAddress, value)
+        ) {
+          result[js] = "transactionReceipt.contractAddress";
+          continue;
+        }
+
+        break;
+      }
+
+      case "log": {
+        for (const argKey of Object.keys(event.event.args)) {
+          const argValue = event.event.args[argKey];
+
+          if (eq(argValue, value)) {
+            result[js] = `args.${argKey}`;
+          }
+        }
+
+        if (result[js]) continue;
+
+        if (eq(event.event.log.address, value)) {
+          result[js] = "log.address";
+          continue;
+        }
+
+        if (eq(event.event.log.logIndex, value)) {
+          result[js] = "log.logIndex";
+          continue;
+        }
+
+        if (eq(event.event.block.hash, value)) {
+          result[js] = "block.hash";
+          continue;
+        }
+
+        if (eq(event.event.block.number, value)) {
+          result[js] = "block.number";
+          continue;
+        }
+
+        if (eq(event.event.block.timestamp, value)) {
+          result[js] = "block.timestamp";
+          continue;
+        }
+
+        if (eq(event.event.block.miner, value)) {
+          result[js] = "block.miner";
+          continue;
+        }
+
+        if (eq(event.event.transaction.hash, value)) {
+          result[js] = "transaction.hash";
+          continue;
+        }
+
+        if (eq(event.event.transaction.from, value)) {
+          result[js] = "transaction.from";
+          continue;
+        }
+
+        if (
+          event.event.transaction.to &&
+          eq(event.event.transaction.to, value)
+        ) {
+          result[js] = "transaction.to";
+          continue;
+        }
+
+        if (eq(event.event.transaction.transactionIndex, value)) {
+          result[js] = "transaction.transactionIndex";
+          continue;
+        }
+
+        if (
+          event.event.transactionReceipt?.contractAddress &&
+          eq(event.event.transactionReceipt.contractAddress, value)
+        ) {
+          result[js] = "transactionReceipt.contractAddress";
+          continue;
+        }
+
+        break;
+      }
+
+      case "trace": {
+        for (const argKey of Object.keys(event.event.args)) {
+          const argValue = event.event.args[argKey];
+
+          if (eq(argValue, value)) {
+            result[js] = `args.${argKey}`;
+          }
+        }
+
+        if (result[js]) continue;
+
+        if (eq(event.event.trace.from, value)) {
+          result[js] = "trace.from";
+          continue;
+        }
+
+        if (eq(event.event.block.hash, value)) {
+          result[js] = "block.hash";
+          continue;
+        }
+
+        if (eq(event.event.block.number, value)) {
+          result[js] = "block.number";
+          continue;
+        }
+
+        if (eq(event.event.block.timestamp, value)) {
+          result[js] = "block.timestamp";
+          continue;
+        }
+
+        if (eq(event.event.block.miner, value)) {
+          result[js] = "block.miner";
+          continue;
+        }
+
+        if (eq(event.event.transaction.hash, value)) {
+          result[js] = "transaction.hash";
+          continue;
+        }
+
+        if (eq(event.event.transaction.from, value)) {
+          result[js] = "transaction.from";
+          continue;
+        }
+
+        if (
+          event.event.transaction.to &&
+          eq(event.event.transaction.to, value)
+        ) {
+          result[js] = "transaction.to";
+          continue;
+        }
+
+        if (eq(event.event.transaction.transactionIndex, value)) {
+          result[js] = "transaction.transactionIndex";
+          continue;
+        }
+
+        if (
+          event.event.transactionReceipt?.contractAddress &&
+          eq(event.event.transactionReceipt.contractAddress, value)
+        ) {
+          result[js] = "transactionReceipt.contractAddress";
+          continue;
+        }
+
+        break;
+      }
+
+      case "transfer": {
+        if (eq(event.event.transfer.from, value)) {
+          result[js] = "transfer.from";
+          continue;
+        }
+
+        if (eq(event.event.transfer.to, value)) {
+          result[js] = "transfer.to";
+          continue;
+        }
+
+        if (eq(event.event.trace.from, value)) {
+          result[js] = "trace.from";
+          continue;
+        }
+
+        if (eq(event.event.block.hash, value)) {
+          result[js] = "block.hash";
+          continue;
+        }
+
+        if (eq(event.event.block.number, value)) {
+          result[js] = "block.number";
+          continue;
+        }
+
+        if (eq(event.event.block.timestamp, value)) {
+          result[js] = "block.timestamp";
+          continue;
+        }
+
+        if (eq(event.event.block.miner, value)) {
+          result[js] = "block.miner";
+          continue;
+        }
+
+        if (eq(event.event.transaction.hash, value)) {
+          result[js] = "transaction.hash";
+          continue;
+        }
+
+        if (eq(event.event.transaction.from, value)) {
+          result[js] = "transaction.from";
+          continue;
+        }
+
+        if (
+          event.event.transaction.to &&
+          eq(event.event.transaction.to, value)
+        ) {
+          result[js] = "transaction.to";
+          continue;
+        }
+
+        if (eq(event.event.transaction.transactionIndex, value)) {
+          result[js] = "transaction.transactionIndex";
+          continue;
+        }
+
+        if (
+          event.event.transactionReceipt?.contractAddress &&
+          eq(event.event.transactionReceipt.contractAddress, value)
+        ) {
+          result[js] = "transactionReceipt.contractAddress";
+          continue;
+        }
+
+        break;
+      }
+    }
+    return undefined;
+  }
+
+  return result;
+};
+
+export const recoverProfilePattern = (
+  pattern: ProfilePattern,
+  event: Event,
+): Row => {
+  const recover = (obj: object, path: string[]): unknown => {
+    if (path.length === 0) return obj;
+    const p = path.splice(0, 1);
+    // @ts-ignore
+    return recover(obj[p[0]], path);
+  };
+
+  const result: Row = {};
+
+  for (const [key, value] of Object.entries(pattern)) {
+    if (key === "chainId") {
+      result[key] = event.chainId;
+    } else {
+      result[key] = recover(event.event, value.split("."));
+    }
+  }
+
+  return result;
+};
diff --git a/packages/core/src/indexing-store/realtime.ts b/packages/core/src/indexing-store/realtime.ts
index 8c4f9960..12a4169f 100644
--- a/packages/core/src/indexing-store/realtime.ts
+++ b/packages/core/src/indexing-store/realtime.ts
@@ -308,6 +308,5 @@ export const createRealtimeIndexingStore = ({
       },
       { schema, casing: "snake_case" },
     ),
-    event: undefined,
   };
 };
diff --git a/packages/core/src/indexing/index.ts b/packages/core/src/indexing/index.ts
index 843911d4..ba0daf53 100644
--- a/packages/core/src/indexing/index.ts
+++ b/packages/core/src/indexing/index.ts
@@ -1,3 +1,4 @@
+import type { IndexingCache } from "@/indexing-store/cache.js";
 import type { IndexingStore } from "@/indexing-store/index.js";
 import type { Common } from "@/internal/common.js";
 import { ShutdownError } from "@/internal/errors.js";
@@ -53,9 +54,11 @@ export type Indexing = {
   processEvents: ({
     events,
     db,
-  }: { events: Event[]; db: IndexingStore }) => Promise<
+    cache,
+  }: { events: Event[]; db: IndexingStore; cache?: IndexingCache }) => Promise<
     { status: "error"; error: Error } | { status: "success" }
   >;
+  getEventCount: () => { [eventName: string]: number };
 };
 
 export const createIndexing = ({
@@ -310,11 +313,13 @@ export const createIndexing = ({
 
       return { status: "success" };
     },
-    async processEvents({ events, db }) {
+    async processEvents({ events, db, cache }) {
       context.db = db;
       for (let i = 0; i < events.length; i++) {
         const event = events[i]!;
-        db.event = event;
+        if (cache) {
+          cache.event = event;
+        }
 
         eventCount[event.name]!++;
 
@@ -339,6 +344,9 @@ export const createIndexing = ({
 
       return { status: "success" };
     },
+    getEventCount() {
+      return eventCount;
+    },
   };
 };
 
diff --git a/packages/core/src/internal/options.ts b/packages/core/src/internal/options.ts
index 4ea74944..b82a3676 100644
--- a/packages/core/src/internal/options.ts
+++ b/packages/core/src/internal/options.ts
@@ -34,12 +34,10 @@ export type Options = {
   factoryAddressCountThreshold: number;
 
   indexingCacheMaxBytes: number;
-  indexingCacheEvictRatio: number;
 
   rpcMaxConcurrency: number;
 
   syncEventsQuerySize: number;
-  syncHandoffStaleSeconds: number;
 };
 
 export const buildOptions = ({ cliOptions }: { cliOptions: CliOptions }) => {
@@ -109,27 +107,17 @@ export const buildOptions = ({ cliOptions }: { cliOptions: CliOptions }) => {
 
     rpcMaxConcurrency: 256,
 
-    // v8.getHeapStatistics().heap_size_limit / 8, bucketed closest to 128, 256, 512, 1024, 2048 mB
+    // v8.getHeapStatistics().heap_size_limit / 5, rounded up to the nearest 64 MB
     indexingCacheMaxBytes:
       process.env.PONDER_CACHE_BYTES !== undefined
         ? Number(process.env.PONDER_CACHE_BYTES)
-        : 2 **
-            Math.min(
-              Math.max(
-                Math.round(
-                  Math.log2(
-                    v8.getHeapStatistics().heap_size_limit / 1_024 / 1_024 / 8,
-                  ),
-                ),
-                7,
-              ),
-              11,
-            ) *
+        : Math.ceil(
+            v8.getHeapStatistics().heap_size_limit / 1_024 / 1_024 / 5 / 64,
+          ) *
+          64 *
           1_024 *
           1_024,
-    indexingCacheEvictRatio: 0.35,
 
     syncEventsQuerySize: 10_000,
-    syncHandoffStaleSeconds: 300,
   } satisfies Options;
 };
diff --git a/packages/core/src/sync-historical/index.ts b/packages/core/src/sync-historical/index.ts
index 8bc042e7..2ca743db 100644
--- a/packages/core/src/sync-historical/index.ts
+++ b/packages/core/src/sync-historical/index.ts
@@ -471,16 +471,34 @@ export const createHistoricalSync = async (
   ////////
 
   const syncLogFilter = async (filter: LogFilter, interval: Interval) => {
-    // Resolve `filter.address`
-    const address = isAddressFactory(filter.address)
-      ? await syncAddressFactory(filter.address, interval).then((result) =>
-          result.size >= args.common.options.factoryAddressCountThreshold
+    let logs: SyncLog[];
+    if (isAddressFactory(filter.address)) {
+      const childAddresses = await syncAddressFactory(filter.address, interval);
+      logs = await syncLogsDynamic({
+        filter,
+        interval,
+        address:
+          childAddresses.size >=
+          args.common.options.factoryAddressCountThreshold
             ? undefined
-            : Array.from(result.keys()),
-        )
-      : filter.address;
+            : Array.from(childAddresses.keys()),
+      });
+
+      logs = logs.filter((log) =>
+        isAddressMatched({
+          address: log.address,
+          blockNumber: hexToNumber(log.blockNumber),
+          childAddresses,
+        }),
+      );
+    } else {
+      logs = await syncLogsDynamic({
+        filter,
+        interval,
+        address: filter.address,
+      });
+    }
 
-    const logs = await syncLogsDynamic({ filter, interval, address });
     await args.syncStore.insertLogs({ logs, chainId: args.network.chainId });
 
     const blocks = await Promise.all(
diff --git a/packages/create-ponder/CHANGELOG.md b/packages/create-ponder/CHANGELOG.md
index 0e994ad9..62aa8c0b 100644
--- a/packages/create-ponder/CHANGELOG.md
+++ b/packages/create-ponder/CHANGELOG.md
@@ -1,5 +1,7 @@
 # create-ponder
 
+## 0.10.7
+
 ## 0.10.6
 
 ## 0.10.5
diff --git a/packages/create-ponder/package.json b/packages/create-ponder/package.json
index abf7582f..e2a52afa 100644
--- a/packages/create-ponder/package.json
+++ b/packages/create-ponder/package.json
@@ -1,6 +1,6 @@
 {
   "name": "create-ponder",
-  "version": "0.10.6",
+  "version": "0.10.7",
   "type": "module",
   "description": "A CLI tool to create Ponder apps",
   "license": "MIT",
diff --git a/packages/eslint-config-ponder/CHANGELOG.md b/packages/eslint-config-ponder/CHANGELOG.md
index 7a5633cc..a434b3ae 100644
--- a/packages/eslint-config-ponder/CHANGELOG.md
+++ b/packages/eslint-config-ponder/CHANGELOG.md
@@ -1,5 +1,7 @@
 # eslint-config-ponder
 
+## 0.10.7
+
 ## 0.10.6
 
 ## 0.10.5
diff --git a/packages/eslint-config-ponder/package.json b/packages/eslint-config-ponder/package.json
index a297b05e..d6520eb6 100644
--- a/packages/eslint-config-ponder/package.json
+++ b/packages/eslint-config-ponder/package.json
@@ -1,6 +1,6 @@
 {
   "name": "eslint-config-ponder",
-  "version": "0.10.6",
+  "version": "0.10.7",
   "description": "ESLint config for Ponder apps",
   "license": "MIT",
   "main": "./index.js",
diff --git a/packages/react/CHANGELOG.md b/packages/react/CHANGELOG.md
index 8a8a1e1f..8f0b12aa 100644
--- a/packages/react/CHANGELOG.md
+++ b/packages/react/CHANGELOG.md
@@ -1,5 +1,12 @@
 # @ponder/react
 
+## 0.10.7
+
+### Patch Changes
+
+- Updated dependencies []:
+  - @ponder/client@0.10.7
+
 ## 0.10.6
 
 ### Patch Changes
diff --git a/packages/react/package.json b/packages/react/package.json
index 3ee70f13..c9ef70bb 100644
--- a/packages/react/package.json
+++ b/packages/react/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@ponder/react",
-  "version": "0.10.6",
+  "version": "0.10.7",
   "description": "React hooks for Ponder",
   "license": "MIT",
   "repository": {
@@ -37,7 +37,7 @@
     "superjson": "^2.2.2"
   },
   "peerDependencies": {
-    "@ponder/client": ">=0.10.6",
+    "@ponder/client": ">=0.10.7",
     "@tanstack/react-query": ">=5.0.0",
     "react": ">=18",
     "typescript": ">=5.0.4"


Test patch:
diff --git a/packages/core/src/indexing-store/cache.test.ts b/packages/core/src/indexing-store/cache.test.ts
index 4d8c1378..44dcb473 100644
--- a/packages/core/src/indexing-store/cache.test.ts
+++ b/packages/core/src/indexing-store/cache.test.ts
@@ -1,3 +1,4 @@
+import { ALICE, BOB } from "@/_test/constants.js";
 import {
   setupCleanup,
   setupCommon,
@@ -5,8 +6,9 @@ import {
   setupIsolatedDatabase,
 } from "@/_test/setup.js";
 import { onchainEnum, onchainTable } from "@/drizzle/onchain.js";
+import type { LogEvent } from "@/internal/types.js";
 import { ZERO_CHECKPOINT_STRING } from "@/utils/checkpoint.js";
-import { zeroAddress } from "viem";
+import { parseEther, zeroAddress } from "viem";
 import { beforeEach, expect, test } from "vitest";
 import { createIndexingCache } from "./cache.js";
 import { createHistoricalIndexingStore } from "./historical.js";
@@ -30,7 +32,7 @@ test("flush() insert", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -75,13 +77,12 @@ test("flush() update", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
     const indexingStore = createHistoricalIndexingStore({
       common: context.common,
-
       schemaBuild: { schema },
       indexingCache,
       db: tx,
@@ -101,13 +102,14 @@ test("flush() update", async (context) => {
       balance: 10n,
     });
 
+    // first flush takes "insert" path
     await indexingCache.flush({ client });
-    indexingCache.commit();
 
     await indexingStore.update(schema.account, { address: zeroAddress }).set({
       balance: 12n,
     });
 
+    // second flush takes "update" path
     await indexingCache.flush({ client });
 
     let result = await indexingStore.find(schema.account, {
@@ -159,7 +161,7 @@ test("flush() encoding", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -221,7 +223,7 @@ test("flush() encoding escape", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -265,7 +267,7 @@ test("flush() encoding escape", async (context) => {
   });
 });
 
-test("commit() evicts rows", async (context) => {
+test("prefetch() queries predicted rows", async (context) => {
   const schema = {
     account: onchainTable("account", (p) => ({
       address: p.hex().primaryKey(),
@@ -277,13 +279,88 @@ test("commit() evicts rows", async (context) => {
     schemaBuild: { schema },
   });
 
+  const event = {
+    type: "log",
+    chainId: 1,
+    checkpoint: ZERO_CHECKPOINT_STRING,
+    name: "Contract:Event",
+    event: {
+      id: ZERO_CHECKPOINT_STRING,
+      args: {
+        from: zeroAddress,
+        to: ALICE,
+        amount: parseEther("1"),
+      },
+      log: {} as LogEvent["event"]["log"],
+      block: {} as LogEvent["event"]["block"],
+      transaction: {} as LogEvent["event"]["transaction"],
+    },
+  } satisfies LogEvent;
+
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
+  });
+
+  indexingCache.event = event;
+
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    await indexingStore
+      .insert(schema.account)
+      .values({
+        address: ALICE,
+        balance: parseEther("1"),
+      })
+      .onConflictDoNothing();
+
+    // @ts-ignore
+    event.event.args.to = BOB;
+
+    await indexingCache.flush({ client });
+    await indexingCache.prefetch({
+      events: [event],
+      db: tx,
+      eventCount: { "Contract:Event": 1 },
+    });
+
+    const result = indexingCache.has({
+      table: schema.account,
+      key: { address: BOB },
+    });
+
+    expect(result).toBe(true);
+  });
+});
+
+test("prefetch() evicts rows", async (context) => {
+  const schema = {
+    account: onchainTable("account", (p) => ({
+      address: p.hex().primaryKey(),
+      balance: p.bigint().notNull(),
+    })),
+  };
+
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  context.common.options.indexingCacheMaxBytes = 0;
+  // skip hot loop
+  indexingCache.invalidate();
 
   await database.transaction(async (client, tx) => {
     const indexingStore = createHistoricalIndexingStore({
@@ -300,7 +377,8 @@ test("commit() evicts rows", async (context) => {
     });
 
     await indexingCache.flush({ client });
-    indexingCache.commit();
+    // prefetch() should evict rows from the cache to free memory
+    await indexingCache.prefetch({ events: [], db: tx, eventCount: {} });
 
     const result = indexingCache.has({
       table: schema.account,
diff --git a/packages/core/src/indexing-store/historical.test.ts b/packages/core/src/indexing-store/historical.test.ts
index 8081c510..4dc78b74 100644
--- a/packages/core/src/indexing-store/historical.test.ts
+++ b/packages/core/src/indexing-store/historical.test.ts
@@ -37,7 +37,7 @@ test("find", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -95,7 +95,7 @@ test("insert", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -249,7 +249,7 @@ test("update", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -343,7 +343,7 @@ test("delete", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -401,7 +401,7 @@ test("sql", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -479,7 +479,7 @@ test("sql followed by find", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -519,7 +519,7 @@ test("onchain table", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -554,7 +554,7 @@ test("missing rows", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -591,7 +591,7 @@ test("notNull", async (context) => {
   let indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -640,7 +640,7 @@ test("notNull", async (context) => {
     indexingCache = createIndexingCache({
       common: context.common,
       schemaBuild: { schema },
-      checkpoint: ZERO_CHECKPOINT_STRING,
+      crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
     });
 
     indexingStore = createHistoricalIndexingStore({
@@ -680,7 +680,7 @@ test("default", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -716,7 +716,7 @@ test("$default", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -755,7 +755,7 @@ test("$onUpdateFn", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -797,7 +797,7 @@ test("array", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -841,7 +841,7 @@ test("text array", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -890,7 +890,7 @@ test("enum", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -933,7 +933,7 @@ test("json bigint", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -967,7 +967,7 @@ test("bytes", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -1008,7 +1008,7 @@ test("text with null bytes", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -1049,7 +1049,7 @@ test.skip("time", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -1090,7 +1090,7 @@ test("timestamp", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -1131,7 +1131,7 @@ test.skip("date", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -1172,7 +1172,7 @@ test.skip("interval", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -1213,7 +1213,7 @@ test("point", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
@@ -1254,7 +1254,7 @@ test("line", async (context) => {
   const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
+    crashRecoveryCheckpoint: ZERO_CHECKPOINT_STRING,
   });
 
   await database.transaction(async (client, tx) => {
diff --git a/packages/core/src/indexing-store/profile.test.ts b/packages/core/src/indexing-store/profile.test.ts
new file mode 100644
index 00000000..de84ec72
--- /dev/null
+++ b/packages/core/src/indexing-store/profile.test.ts
@@ -0,0 +1,206 @@
+import { onchainTable } from "@/drizzle/onchain.js";
+import type { BlockEvent, LogEvent } from "@/internal/types.js";
+import { ZERO_CHECKPOINT_STRING } from "@/utils/checkpoint.js";
+import type { Column, Table } from "drizzle-orm";
+import { zeroAddress } from "viem";
+import { expect, test } from "vitest";
+import { recordProfilePattern } from "./profile.js";
+
+test("recordProfilePattern() no pattern", () => {
+  const event = {
+    type: "block",
+    chainId: 1,
+    checkpoint: ZERO_CHECKPOINT_STRING,
+    name: "",
+    event: {
+      id: ZERO_CHECKPOINT_STRING,
+      block: {} as BlockEvent["event"]["block"],
+    },
+  } satisfies BlockEvent;
+
+  const schema = {
+    account: onchainTable("account", (p) => ({
+      address: p.hex().primaryKey(),
+      balance: p.bigint().notNull(),
+    })),
+  };
+
+  const primaryKeyCache = new Map<Table, [string, Column][]>();
+
+  primaryKeyCache.set(schema.account, [["address", schema.account.address]]);
+
+  const pattern = recordProfilePattern(
+    event,
+    schema.account,
+    { address: zeroAddress },
+    [],
+    primaryKeyCache,
+  );
+
+  expect(pattern).toBeUndefined();
+});
+
+test("recordProfilePattern() chainId", () => {
+  const event = {
+    type: "block",
+    chainId: 1,
+    checkpoint: ZERO_CHECKPOINT_STRING,
+    name: "",
+    event: {
+      id: ZERO_CHECKPOINT_STRING,
+      block: {} as BlockEvent["event"]["block"],
+    },
+  } satisfies BlockEvent;
+
+  const schema = {
+    account: onchainTable("account", (p) => ({
+      address: p.integer().primaryKey(),
+      balance: p.bigint().notNull(),
+    })),
+  };
+
+  const primaryKeyCache = new Map<Table, [string, Column][]>();
+
+  primaryKeyCache.set(schema.account, [["address", schema.account.address]]);
+
+  const pattern = recordProfilePattern(
+    event,
+    schema.account,
+    { address: 1 },
+    [],
+    primaryKeyCache,
+  );
+
+  expect(pattern).toMatchInlineSnapshot(`
+    {
+      "address": "chainId",
+    }
+  `);
+});
+
+test("recordProfilePattern() log args", () => {
+  const event = {
+    type: "log",
+    chainId: 1,
+    checkpoint: ZERO_CHECKPOINT_STRING,
+    name: "",
+    event: {
+      id: ZERO_CHECKPOINT_STRING,
+      args: {
+        address: zeroAddress,
+      },
+      log: {} as LogEvent["event"]["log"],
+      transaction: {} as LogEvent["event"]["transaction"],
+      block: {} as BlockEvent["event"]["block"],
+    },
+  } satisfies LogEvent;
+
+  const schema = {
+    account: onchainTable("account", (p) => ({
+      address: p.hex().primaryKey(),
+      balance: p.bigint().notNull(),
+    })),
+  };
+
+  const primaryKeyCache = new Map<Table, [string, Column][]>();
+
+  primaryKeyCache.set(schema.account, [["address", schema.account.address]]);
+
+  const pattern = recordProfilePattern(
+    event,
+    schema.account,
+    { address: zeroAddress },
+    [],
+    primaryKeyCache,
+  );
+
+  expect(pattern).toMatchInlineSnapshot(`
+    {
+      "address": "args.address",
+    }
+  `);
+});
+
+test("recordProfilePattern() block", () => {
+  const event = {
+    type: "block",
+    chainId: 1,
+    checkpoint: ZERO_CHECKPOINT_STRING,
+    name: "",
+    event: {
+      id: ZERO_CHECKPOINT_STRING,
+      block: { number: 3n } as BlockEvent["event"]["block"],
+    },
+  } satisfies BlockEvent;
+
+  const schema = {
+    account: onchainTable("account", (p) => ({
+      address: p.bigint().primaryKey(),
+      balance: p.bigint().notNull(),
+    })),
+  };
+
+  const primaryKeyCache = new Map<Table, [string, Column][]>();
+
+  primaryKeyCache.set(schema.account, [["address", schema.account.address]]);
+
+  const pattern = recordProfilePattern(
+    event,
+    schema.account,
+    { address: 3n },
+    [],
+    primaryKeyCache,
+  );
+
+  expect(pattern).toMatchInlineSnapshot(`
+    {
+      "address": "block.number",
+    }
+  `);
+});
+
+test("recordProfilePattern() hint", () => {
+  const event = {
+    type: "block",
+    chainId: 1,
+    checkpoint: ZERO_CHECKPOINT_STRING,
+    name: "",
+    event: {
+      id: ZERO_CHECKPOINT_STRING,
+      block: { number: 3n } as BlockEvent["event"]["block"],
+    },
+  } satisfies BlockEvent;
+
+  const schema = {
+    account: onchainTable("account", (p) => ({
+      address: p.bigint().primaryKey(),
+      balance: p.bigint().notNull(),
+    })),
+  };
+
+  const primaryKeyCache = new Map<Table, [string, Column][]>();
+
+  primaryKeyCache.set(schema.account, [["address", schema.account.address]]);
+
+  let pattern = recordProfilePattern(
+    event,
+    schema.account,
+    { address: 3n },
+    [],
+    primaryKeyCache,
+  );
+
+  pattern = recordProfilePattern(
+    event,
+    schema.account,
+    { address: 3n },
+    [pattern!],
+    primaryKeyCache,
+  );
+
+  expect(pattern).toMatchInlineSnapshot(`
+      {
+        "address": "block.number",
+      }
+    `);
+});