fix: Eliminate top DB queries — replace hourly_statistics MV with aggregate table, batch balance lookups, fix VACUUM in transaction

nelitow · nelitow · commit a44393e35cf4 · 2026-03-13T15:43:26.000-03:00
- Replace hourly_statistics MV with incremental aggregate table (migration 046):
  - MV refresh was top query at 201,290ms latency, running every 30min
  - Per-block upsert recomputes only the affected hour (~25ms vs 2,132ms locally)
  - 85x improvement; cost scales with 1 hour of data not 24h history
- Add composite index (account_hash, asset_id, _id DESC) on balance table (migration 047):
  - Enables single index seek for IndexBalance lookups instead of multi-index scan
- Batch balance lookups in IndexBalance: 1 query per block instead of N+1 per (account, asset) pair
- Fix queryWithTimeout: VACUUM and REFRESH MATERIALIZED VIEW CONCURRENTLY cannot
  run inside a transaction block — detect these and use session-level SET instead of SET LOCAL
diff --git a/packages/graphql/database/migrations/046_replace_hourly_statistics_mv_with_agg_table.sql b/packages/graphql/database/migrations/046_replace_hourly_statistics_mv_with_agg_table.sql
@@ -0,0 +1,41 @@
+-- Migration 046: Replace hourly_statistics MV with aggregate table
+--
+-- Problem: REFRESH MATERIALIZED VIEW was the top DB query at 201,290ms latency
+-- It does a full seq scan of all 24h transactions every 30 minutes.
+--
+-- Solution: Incremental aggregate table updated per-batch during indexing.
+-- Each batch recomputes only the affected hours (typically 1-2 hours),
+-- reducing cost from full 24h scan to a narrow timestamp index scan.
+--
+-- Benchmark (local, 17k blocks):
+--   Before: 2,132ms (full seq scan + disk sort)
+--   After:  ~25ms   (index scan on timestamp for affected hours only)
+
+-- Create the aggregate table
+CREATE TABLE IF NOT EXISTS indexer.hourly_statistics_agg (
+  hour        TIMESTAMPTZ NOT NULL,
+  total_fee   NUMERIC     NOT NULL DEFAULT 0,
+  total_gas_used BIGINT   NOT NULL DEFAULT 0,
+  PRIMARY KEY (hour)
+);
+
+-- Grant read access
+GRANT SELECT ON indexer.hourly_statistics_agg TO explorer_ro;
+
+-- Backfill from existing MV if it has data
+INSERT INTO indexer.hourly_statistics_agg (hour, total_fee, total_gas_used)
+SELECT hour, total_fee, total_gas_used
+FROM indexer.hourly_statistics
+ON CONFLICT (hour) DO UPDATE
+  SET total_fee = excluded.total_fee,
+      total_gas_used = excluded.total_gas_used;
+
+-- Remove the old MV refresh jobs (query strings include SET/RESET statement_timeout wrappers)
+DELETE FROM indexer.database_jobs
+WHERE query LIKE '%REFRESH MATERIALIZED VIEW CONCURRENTLY indexer.hourly_statistics%';
+
+DELETE FROM indexer.database_jobs
+WHERE query LIKE '%ANALYZE indexer.hourly_statistics%';
+
+-- Drop the old MV
+DROP MATERIALIZED VIEW IF EXISTS indexer.hourly_statistics CASCADE;
diff --git a/packages/graphql/database/migrations/047_optimize_balance_index.sql b/packages/graphql/database/migrations/047_optimize_balance_index.sql
@@ -0,0 +1,12 @@
+-- Migration 047: Add composite index for balance lookups
+--
+-- Problem: IndexBalance queries `select balance ... where account_hash = $1 and asset_id = $2
+-- order by _id desc limit 1` at 43.27 calls/sec (top DB load query).
+-- Separate single-column indexes on account_hash and asset_id force PostgreSQL
+-- to use one index and filter/sort the rest, leading to excessive I/O.
+--
+-- Solution: Composite index on (account_hash, asset_id, _id DESC) so the query
+-- becomes a single index seek with no post-filter sort.
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS balance_account_asset_id_idx
+ON indexer.balance (account_hash, asset_id, _id DESC);
diff --git a/packages/graphql/src/application/uc/IndexBalance.ts b/packages/graphql/src/application/uc/IndexBalance.ts
@@ -23,6 +23,55 @@ export default class IndexBalance {
         'delete from indexer.balance where block_height = $1',
         [block.height],
       );
+
+      // Collect all (account, asset) pairs across all transactions in this block
+      const pairs: { accountHash: string; assetId: string }[] = [];
+      for (const transaction of block.transactions) {
+        transaction.inputs = transaction.inputs || [];
+        transaction.outputs = transaction.outputs || [];
+        for (const input of transaction.inputs) {
+          if (input.__typename === 'InputCoin') {
+            pairs.push({ accountHash: input.owner, assetId: input.assetId });
+          }
+        }
+        for (const output of transaction.outputs) {
+          if (
+            ['ChangeOutput', 'CoinOutput', 'VariableOutput'].includes(
+              output.__typename,
+            )
+          ) {
+            pairs.push({ accountHash: output.to, assetId: output.assetId });
+          }
+        }
+      }
+
+      // Batch fetch latest balance for all (account, asset) pairs in one query
+      const balanceMap = new Map<string, BigNumber>();
+      if (pairs.length > 0) {
+        const uniquePairs = [
+          ...new Map(
+            pairs.map((p) => [`${p.accountHash}:${p.assetId}`, p]),
+          ).values(),
+        ];
+        const placeholders = uniquePairs
+          .map((_, i) => `($${i * 2 + 1}, $${i * 2 + 2})`)
+          .join(', ');
+        const params = uniquePairs.flatMap((p) => [p.accountHash, p.assetId]);
+        const rows = await connection.query(
+          `SELECT DISTINCT ON (account_hash, asset_id) account_hash, asset_id, balance
+           FROM indexer.balance
+           WHERE (account_hash, asset_id) IN (${placeholders})
+           ORDER BY account_hash, asset_id, _id DESC`,
+          params,
+        );
+        for (const row of rows) {
+          balanceMap.set(
+            `${row.account_hash}:${row.asset_id}`,
+            BigNumber(row.balance),
+          );
+        }
+      }
+
       for (const transaction of block.transactions) {
         transaction.inputs = transaction.inputs || [];
         transaction.outputs = transaction.outputs || [];
@@ -61,11 +110,8 @@ export default class IndexBalance {
         }
         for (const accountHash in index) {
           for (const assetId in index[accountHash]) {
-            const [balanceData] = await connection.query(
-              'select balance from indexer.balance where account_hash = $1 and asset_id = $2 order by _id desc limit 1',
-              [accountHash, assetId],
-            );
-            let balance = BigNumber(balanceData ? balanceData.balance : 0);
+            const key = `${accountHash}:${assetId}`;
+            let balance = balanceMap.get(key) ?? BigNumber(0);
             const events = index[accountHash][assetId];
             for (const event of events) {
               if (event.type === 'input') {
@@ -75,6 +121,8 @@ export default class IndexBalance {
                 balance = balance.plus(BigNumber(event.amount));
               }
             }
+            // Update map so subsequent transactions in same block see updated balance
+            balanceMap.set(key, balance);
             await connection.query(
               'insert into indexer.balance (block_height, tx_hash, account_hash, asset_id, balance) values ($1, $2, $3, $4, $5) on conflict do nothing',
               [
diff --git a/packages/graphql/src/application/uc/NewAddBlockRange.ts b/packages/graphql/src/application/uc/NewAddBlockRange.ts
@@ -62,6 +62,25 @@ export default class NewAddBlockRange {
           block.data.header.daHeight,
         ],
       });
+      // Incrementally update hourly_statistics_agg for this block's hour.
+      // Using DELETE + INSERT pattern for idempotency: recomputes the entire
+      // hour from indexer.transactions so re-processing a block never double-counts.
+      queries.push({
+        statement: `
+          INSERT INTO indexer.hourly_statistics_agg (hour, total_fee, total_gas_used)
+          SELECT
+            date_trunc('hour', $1::timestamptz) AS hour,
+            COALESCE(SUM((data->'status'->>'totalFee')::numeric), 0) AS total_fee,
+            COALESCE(SUM((data->'status'->>'gasUsed')::bigint), 0) AS total_gas_used
+          FROM indexer.transactions
+          WHERE timestamp >= date_trunc('hour', $1::timestamptz)
+            AND timestamp < date_trunc('hour', $1::timestamptz) + INTERVAL '1 hour'
+          ON CONFLICT (hour) DO UPDATE
+            SET total_fee = excluded.total_fee,
+                total_gas_used = excluded.total_gas_used
+        `,
+        params: [block.timestamp],
+      });
       for (const [index, transactionData] of blockData.transactions.entries()) {
         const transaction = new Transaction(
           transactionData,
diff --git a/packages/graphql/src/infra/dao/BlockDAO.ts b/packages/graphql/src/infra/dao/BlockDAO.ts
@@ -378,7 +378,7 @@ export default class BlockDAO {
     const data = await this.databaseConnection.query(
       `SELECT to_char(hour, 'YYYY-MM-DD HH24') AS date,
         total_fee AS value
-      FROM indexer.hourly_statistics
+      FROM indexer.hourly_statistics_agg
       WHERE hour > NOW() - INTERVAL '24 hours'
         AND hour <= date_trunc('hour', NOW())
       ORDER BY hour`,
diff --git a/packages/graphql/src/infra/database/DatabaseConnection.ts b/packages/graphql/src/infra/database/DatabaseConnection.ts
@@ -54,18 +54,33 @@ export class DatabaseConnection {
 
   /**
    * Execute a query with per-session statement_timeout and lock_timeout.
-   * Timeouts are applied via SET LOCAL inside a transaction so they don't
-   * leak to other connections. If the query exceeds the timeout, PostgreSQL
-   * cancels it and the connection is returned cleanly to the pool.
+   *
+   * REFRESH MATERIALIZED VIEW CONCURRENTLY cannot run inside a transaction block.
+   * For those queries, timeouts are applied as session-level SET commands and
+   * reset afterward. For all other queries, SET LOCAL inside BEGIN/COMMIT is used
+   * so timeouts don't leak to other connections.
    */
   async queryWithTimeout(
     statement: string,
     params: any,
     statementTimeoutMs: number,
     lockTimeoutMs: number,
   ) {
+    // VACUUM and REFRESH MATERIALIZED VIEW CONCURRENTLY cannot run inside a transaction block
+    const requiresNoTransaction =
+      /REFRESH\s+MATERIALIZED\s+VIEW\s+CONCURRENTLY/i.test(statement) ||
+      /^\s*VACUUM\b/i.test(statement);
     const connection = await this.pool.connect();
     try {
+      if (requiresNoTransaction) {
+        // Cannot use BEGIN/COMMIT — apply timeouts at session level and reset after
+        await connection.query(`SET statement_timeout = ${statementTimeoutMs}`);
+        await connection.query(`SET lock_timeout = ${lockTimeoutMs}`);
+        const result = await connection.query(statement, params);
+        await connection.query('RESET statement_timeout');
+        await connection.query('RESET lock_timeout');
+        return result.rows;
+      }
       await connection.query('BEGIN');
       await connection.query(
         `SET LOCAL statement_timeout = ${statementTimeoutMs}`,
@@ -75,9 +90,16 @@ export class DatabaseConnection {
       await connection.query('COMMIT');
       return result.rows;
     } catch (error) {
-      try {
-        await connection.query('ROLLBACK');
-      } catch {}
+      if (!requiresNoTransaction) {
+        try {
+          await connection.query('ROLLBACK');
+        } catch {}
+      } else {
+        try {
+          await connection.query('RESET statement_timeout');
+          await connection.query('RESET lock_timeout');
+        } catch {}
+      }
       throw error;
     } finally {
       connection.release();
