perf: added two new variables to optimize memory consumption during reindex.

> Products limit chunk size for each loop iteration when indexing products.
`reindexProductsChunkSize` - default value is 500

> Index operations are performed in bulk, with each bulk operation containing a number of individualindex operations. This option sets the maximum number of operations in the memory buffer before abulk operation is executed.
`reindexBulkOperationSizeLimit` - default value is 3000

Refs: https://github.com/vendure-ecommerce/vendure/issues/1770

docs/content/typescript-api/core-plugins/elasticsearch-plugin/elasticsearch-options.md

@@ -166,13 +166,23 @@ indexMappingProperties: {
     type: 'integer',
   }
 }
-```{{< /member-description >}}
+```
+
+{{< /member-description >}}
+
+### reindexProductChunkSize
+
+{{< member-info kind="property" type="number" default="500"  >}}
+
+{{< member-description >}}Products limit chunk size for each loop iteration when indexing products.{{< /member-description >}}
 
-### batchSize
+### reindexBulkOperationSizeLimit
 
-{{< member-info kind="property" type="number" default="2000"  >}}
+{{< member-info kind="property" type="number" default="3000"  >}}
 
-{{< member-description >}}Batch size for bulk operations (e.g. when rebuilding the indices).{{< /member-description >}}
+{{< member-description >}}Index operations are performed in bulk, with each bulk operation 
+containing a number of individualindex operations. This option sets the maximum number of operations 
+in the memory buffer before abulk operation is executed.{{< /member-description >}}
 
 ### searchConfig
 

packages/elasticsearch-plugin/src/indexing/indexer.controller.ts

@@ -258,6 +258,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                 }
 
                 const totalProductIds = await this.connection
+                    .rawConnection
                     .getRepository(Product)
                     .createQueryBuilder('product')
                     .where('product.deletedAt IS NULL')
@@ -269,7 +270,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                 let skip = 0;
                 let finishedProductsCount = 0;
                 do {
-                    const operations: BulkVariantOperation[] = [];
+                    let operations: BulkVariantOperation[] = [];
 
                     productIds = await this.connection
                         .getRepository(Product)
@@ -277,11 +278,21 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                         .select('product.id')
                         .where('product.deletedAt IS NULL')
                         .skip(skip)
-                        .take(REINDEX_CHUNK_SIZE)
+                        .take(this.options.reindexProductsChunkSize)
                         .getMany();
 
                     for (const { id: productId } of productIds) {
                         operations.push(...(await this.updateProductsOperationsOnly(ctx, productId)));
+                        if (operations.length >= this.options.reindexBulkOperationSizeLimit) {
+                            // Because we can have a huge amount of variant for 1 product, we also chunk update operations
+                            await this.executeBulkOperationsByChunks(
+                                this.options.reindexBulkOperationSizeLimit,
+                                operations,
+                                variantIndexNameForReindex,
+                            );
+                            // Reset operations to avoid memory peaks with huge amount of operations
+                            operations = [];
+                        }
                         finishedProductsCount++;
                         observer.next({
                             total: totalProductIds,
@@ -294,15 +305,15 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
 
                     // Because we can have a huge amount of variant for 1 product, we also chunk update operations
                     await this.executeBulkOperationsByChunks(
-                        REINDEX_OPERATION_CHUNK_SIZE,
+                        this.options.reindexBulkOperationSizeLimit,
                         operations,
                         variantIndexNameForReindex,
                     );
 
-                    skip += REINDEX_CHUNK_SIZE;
+                    skip += this.options.reindexProductsChunkSize;
 
                     Logger.verbose(`Done ${finishedProductsCount} / ${totalProductIds} products`);
-                } while (productIds.length >= REINDEX_CHUNK_SIZE);
+                } while (productIds.length >= this.options.reindexProductsChunkSize);
 
                 // Switch the index to the new reindexed one
                 try {

packages/elasticsearch-plugin/src/options.ts

@@ -150,12 +150,22 @@ export interface ElasticsearchOptions {
     };
     /**
      * @description
-     * Batch size for bulk operations (e.g. when rebuilding the indices).
+     * Products limit chunk size for each loop iteration when indexing products.
      *
      * @default
-     * 2000
+     * 500
      */
-    batchSize?: number;
+    reindexProductsChunkSize?: number;
+    /**
+     * @description
+     * Index operations are performed in bulk, with each bulk operation containing a number of individual
+     * index operations. This option sets the maximum number of operations in the memory buffer before a
+     * bulk operation is executed.
+     *
+     * @default
+     * 3000
+     */
+    reindexBulkOperationSizeLimit?: number;
     /**
      * @description
      * Configuration of the internal Elasticsearch query.
@@ -702,7 +712,8 @@ export const defaultOptions: ElasticsearchRuntimeOptions = {
     indexPrefix: 'vendure-',
     indexSettings: {},
     indexMappingProperties: {},
-    batchSize: 2000,
+    productChunkSize: 500,
+    bulkOperationSizeLimit: 3000,
     searchConfig: {
         facetValueMaxSize: 50,
         collectionMaxSize: 50,