Improve reindex query

packages/node-core/CHANGELOG.md

@@ -7,15 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Fixed
+- Improve reindex query to remove/update by batches. (#2131)
+- Fix poi reindex beyond genesis height issue. (#2131)
 - Fixed modulo block ahead of finalized block issue (#2132)
-
+- Wrong link to docs for testing
 ### Added
 - WorkerInMemoryCacheService from node (#2125)
 - New `endBlock` option on datasources (#2064)
 
-### Fixed
-- Wrong link to docs for testing
-
 ## [6.1.1] - 2023-10-25
 ### Fixed
 - Not handling TaskFlushedErrors in block dispatchers (#2120)

packages/node-core/src/indexer/poi/poi.service.ts

@@ -7,6 +7,7 @@ import {Op, QueryTypes, Transaction} from '@subql/x-sequelize';
 import {NodeConfig} from '../../configure';
 import {getLogger} from '../../logger';
 import {sqlIterator} from '../../utils';
+import {PoiRepo} from '../entities';
 import {StoreCacheService} from '../storeCache';
 import {CachePoiModel} from '../storeCache/cachePoi';
 import {ISubqueryProject} from '../types';
@@ -114,7 +115,7 @@ export class PoiService implements OnApplicationShutdown {
         }
         if (!checkResult.parent_nullable) {
           queries.push(`ALTER TABLE ${tableName} ALTER COLUMN "parentHash" DROP NOT NULL;`);
-          queries.push(sqlIterator(tableName, `UPDATE ${tableName} SET "parentHash" = NULL;`));
+          queries.push(sqlIterator(tableName, `UPDATE ${tableName} SET "parentHash" = NULL`));
 
           queries.push(
             sqlIterator(
@@ -151,18 +152,65 @@ export class PoiService implements OnApplicationShutdown {
   }
 
   async rewind(targetBlockHeight: number, transaction: Transaction): Promise<void> {
-    await this.poiRepo.model.destroy({
-      transaction,
-      where: {
-        id: {
-          [Op.gt]: targetBlockHeight,
-        },
-      },
-    });
+    await batchDeletePoi(this.poiRepo.model, transaction, targetBlockHeight);
     const lastSyncedPoiHeight = await this.storeCache.metadata.find('latestSyncedPoiHeight');
+
     if (lastSyncedPoiHeight !== undefined && lastSyncedPoiHeight > targetBlockHeight) {
-      this.storeCache.metadata.set('latestSyncedPoiHeight', targetBlockHeight);
+      const genesisPoi = await this.poiRepo.model.findOne({
+        order: [['id', 'ASC']],
+        transaction: transaction,
+      });
+      // This indicates reindex height is less than genesis poi height
+      // And genesis poi has been remove from `batchDeletePoi`
+      if (!genesisPoi) {
+        this.storeCache.metadata.bulkRemove(['latestSyncedPoiHeight']);
+      } else {
+        this.storeCache.metadata.set('latestSyncedPoiHeight', targetBlockHeight);
+      }
     }
     this.storeCache.metadata.bulkRemove(['lastCreatedPoiHeight']);
   }
 }
+
+// REMOVE 10,000 record per batch
+async function batchDeletePoi(
+  model: PoiRepo,
+  transaction: Transaction,
+  targetBlockHeight: number,
+  batchSize = 10000
+): Promise<void> {
+  let completed = false;
+  // eslint-disable-next-line no-constant-condition
+  while (!completed) {
+    try {
+      const recordsToDelete = await model.findAll({
+        transaction,
+        limit: batchSize,
+        where: {
+          id: {
+            [Op.gt]: targetBlockHeight,
+          },
+        },
+      });
+      if (recordsToDelete.length === 0) {
+        break;
+        completed = true;
+      }
+      logger.debug(`Found Poi recordsToDelete ${recordsToDelete.length}`);
+      if (recordsToDelete.length) {
+        await model.destroy({
+          transaction,
+          hooks: false,
+          where: {
+            id: {
+              [Op.in]: recordsToDelete.map((record) => record.id),
+            },
+          },
+        });
+      }
+      // offset += batchSize;
+    } catch (e) {
+      throw new Error(`Reindex model Poi failed, please try to reindex again: ${e}`);
+    }
+  }
+}

packages/node-core/src/indexer/poi/poiModel.ts

@@ -61,6 +61,7 @@ export class PlainPoiModel implements PoiInterface {
     }
     await this.model.bulkCreate(proofs, {
       transaction: tx,
+      conflictAttributes: ['id'],
       updateOnDuplicate: ['hash', 'parentHash'],
     });
   }

packages/node-core/src/indexer/store.service.ts

@@ -714,35 +714,14 @@ group by
     if (!this.historical) {
       throw new Error('Unable to reindex, historical state not enabled');
     }
+    // This should only been called from CLI, blockHeight in storeService never been set and is required for`beforeFind` hook
+    // Height no need to change for rewind during indexing
+    if (this._blockHeight === undefined) {
+      this.setBlockHeight(targetBlockHeight);
+    }
     for (const model of Object.values(this.sequelize.models)) {
       if ('__block_range' in model.getAttributes()) {
-        await model.destroy({
-          transaction,
-          hooks: false,
-          where: this.sequelize.where(
-            this.sequelize.fn('lower', this.sequelize.col('_block_range')),
-            Op.gt,
-            targetBlockHeight
-          ),
-        });
-        await model.update(
-          {
-            __block_range: this.sequelize.fn(
-              'int8range',
-              this.sequelize.fn('lower', this.sequelize.col('_block_range')),
-              null
-            ),
-          },
-          {
-            transaction,
-            hooks: false,
-            where: {
-              __block_range: {
-                [Op.contains]: targetBlockHeight,
-              },
-            },
-          }
-        );
+        await batchDeleteAndThenUpdate(this.sequelize, model, transaction, targetBlockHeight);
       }
     }
     this.metadataModel.set('lastProcessedHeight', targetBlockHeight);
@@ -907,3 +886,76 @@ group by
     };
   }
 }
+
+// REMOVE 10,000 record per batch
+async function batchDeleteAndThenUpdate(
+  sequelize: Sequelize,
+  model: ModelStatic<any>,
+  transaction: Transaction,
+  targetBlockHeight: number,
+  batchSize = 10000
+): Promise<void> {
+  let offset = 0;
+  let completed = false;
+  // eslint-disable-next-line no-constant-condition
+  while (!completed) {
+    try {
+      const [recordsToUpdate, recordsToDelete] = await Promise.all([
+        model.findAll({
+          transaction,
+          limit: batchSize,
+          attributes: {include: ['_id']},
+          offset, // We need to apply offset, because after update the records, the record could still with in range, avoid endless query here.
+          where: {
+            __block_range: {
+              [Op.contains]: targetBlockHeight,
+            },
+          },
+        }),
+        model.findAll({
+          transaction,
+          limit: batchSize,
+          attributes: {include: ['_id']},
+          where: sequelize.where(sequelize.fn('lower', sequelize.col('_block_range')), Op.gt, targetBlockHeight),
+        }),
+      ]);
+      if (recordsToDelete.length === 0 && recordsToUpdate.length === 0) {
+        break;
+        completed = true;
+      }
+      logger.debug(
+        `Found ${model.name} recordsToDelete ${recordsToDelete.length},recordsToUpdate ${recordsToUpdate.length}`
+      );
+      if (recordsToDelete.length) {
+        await model.destroy({
+          transaction,
+          hooks: false,
+          where: {
+            _id: {
+              [Op.in]: recordsToDelete.map((record) => record.dataValues._id),
+            },
+          },
+        });
+      }
+      if (recordsToUpdate.length) {
+        await model.update(
+          {
+            __block_range: sequelize.fn('int8range', sequelize.fn('lower', sequelize.col('_block_range')), null),
+          },
+          {
+            transaction,
+            hooks: false,
+            where: {
+              _id: {
+                [Op.in]: recordsToUpdate.map((record) => record.dataValues._id),
+              },
+            },
+          }
+        );
+      }
+      offset += batchSize;
+    } catch (e) {
+      throw new Error(`Reindex update model ${model.name} failed, please try to reindex again: ${e}`);
+    }
+  }
+}