Repository: typedorm/typedorm (ts)

Issue:
Parallel scan support
Ability to perform parallel scans when using scan manager. 
This would require processing all requests in a parallel-like structure over the promise.all API.

# Taks
- [x] scan transformer to support transforming parallel scan items
- [x] scan manager `scan` to support additional concurrency options and perform multiple requests in parallel using promise.all
- [x] scan manager's higher level `.find` and `.count` to facilitate running parallel requests
- [x] add test coverage
- [x] update docs reflecting support for parallel scans

Pull request:
Feature/73 parallel scan support
# Features

- parallel scan support
- find operation now by default runs a parallel scan when asked to find an entity with segments
- improved scan manager option documentation
- improved error handling for scan response

closes #73

Patch:
diff --git a/README.md b/README.md
index ab70473..483eb74 100644
--- a/README.md
+++ b/README.md
@@ -32,9 +32,11 @@ TypeDORM borrows decorator based syntax from TypeORM and provides fully type saf
 - Attribute level per entity transformation, enabled via [class-transformer](https://github.com/typestack/class-transformer)
 - Full type safety
 - Declarative relational schema
-- Entity manager
-- Transaction manager
-- Batch manager
+- Entity manager - easy to use `findOne`, `find`, `count`, `exists`, `create`, `update`, and `delete` operations
+- Transaction manager - easy to use `write` and `read` operations
+- Batch manager - powerful `write` and `read` operations
+- Scan Manager - powerful `find`, `count`, `parallelScan` and `scan` operations
+- Safer parallel scan with configurable concurrency control
 - Multiple connections support
 - Code follows all possible best practices when modeling for dynamodb
 - Supports specifying non key attribute as unique, built in pagination support for querying
diff --git a/docs/api.md b/docs/api.md
index 1dd0acd..01baf36 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -27,6 +27,7 @@ _Note: In an event of inconstancy between actual API and this document, API shou
   - [ScanManager](#scanmanager)
     - [ScanManager.find](#scanmanagerfind)
     - [ScanManager.count](#scanmanagercount)
+    - [ScanManager.parallelScan](#scanmanagerparallelscan)
     - [ScanManager.scan](#scanmanagerscan)
 
 ## Connection
@@ -682,6 +683,20 @@ find(
     // Specify attributes to get, only selected attributes are fetched
     // @default `ALL`
     select
+
+    // @optional
+    // Total segments to divide this scan in
+    totalSegments
+
+    // @optional
+    // Item scan limit to apply per segment, this option is ignored when `totalSegments` is not provided
+    limitPerSegment
+
+    // @optional
+    // maximum concurrency to apply, this option is ignored when `totalSegments` is not provided
+    // @default PARALLEL_SCAN_CONCURRENCY_LIMIT
+    requestConcurrencyLimit
+
   },
 
   // @optional
@@ -740,13 +755,93 @@ count(
 )
 ```
 
+### ScanManager.parallelScan
+
+You should always use `.find` unless trying to scan all items from the table
+
+Returns all scanned items to the client that matches given filter, this is build on top of lower level `.scan` op but adds
+safer concurrency control - also used by `.find`.
+
+```Typescript
+parallelScan(
+  // @optional
+  // Scan manager scan options
+  // @default none
+  scanOptions: {
+    // @Optional
+    // Entity to scan
+    // @default - none
+    entity
+
+    // @optional
+    // Name of the index if running a scan against secondary indexes
+    // @default - scan will be run against main table
+    scanIndex
+
+    // @optional
+    // Total number of items to return
+    // @default no limit is applied
+    limit
+
+    // @optional
+    // Cursor to start scanning from
+    // @default - none
+    cursor
+
+    // @optional
+    // Filter returned items
+    // Any conditions listed here will apply after items have been read from dynamodb and
+    // therefore this should be avoided wherever possible, but can be helpful in some cases
+    // see this https://www.alexdebrie.com/posts/dynamodb-filter-expressions/ for more details
+    where
+
+    // @optional
+    // Specify attributes to get, only selected attributes are fetched
+    // @default `ALL`
+    select
+
+    // @optional
+    // Current segment identified
+    segment
+
+    // @optional
+    // Total segments to divide this scan in
+    totalSegments
+
+    // @optional
+    // Item scan limit to apply per segment, this option is ignored when `totalSegments` is not provided
+    limitPerSegment
+
+    // @optional
+    // maximum concurrency to apply, this option is ignored when `totalSegments` is not provided
+    // @default PARALLEL_SCAN_CONCURRENCY_LIMIT
+    requestConcurrencyLimit
+  },
+
+  // @optional
+  // extra non-functional options
+  metadataOptions: {
+    // @optional
+    // Unique request id to use, throughout the request processing,
+    // @default a unique v4 uuid is set and used for all logs
+    requestId
+
+    // @optional
+    // Sets ReturnConsumedCapacity param to given value when making a request via document client
+    returnConsumedCapacity
+  }
+)
+```
+
 ### ScanManager.scan
 
 Low level scan api - returns all scanned items to the client that matches given filter - also used by `.find`.
-You should always use `.find` unless trying to scan all items from the table,
+You should always use `.find` unless trying to scan all items from the table.
+
+_Scan manager `.scan` does not enforce concurrency limit, consider using `.parallelScan` to perform safer parallel scan._
 
 ```Typescript
-count(
+scan(
   // @optional
   // Scan manager scan options
   // @default none
@@ -782,6 +877,18 @@ count(
     // Specify attributes to get, only selected attributes are fetched
     // @default `ALL`
     select
+
+    // @optional
+    // Current segment identified
+    segment
+
+    // @optional
+    // Total segments to divide this scan in
+    totalSegments
+
+    // @optional
+    // Item scan limit to apply per segment, this option is ignored when `totalSegments` is not provided
+    limitPerSegment
   },
 
   // @optional
diff --git a/packages/common/src/constants.ts b/packages/common/src/constants.ts
index e903387..baefe2b 100644
--- a/packages/common/src/constants.ts
+++ b/packages/common/src/constants.ts
@@ -34,6 +34,11 @@ export const BATCH_READ_ITEMS_LIMIT = 100;
  */
 export const BATCH_WRITE_CONCURRENCY_LIMIT = 5;
 
+/**
+ * Concurrency limit to apply, when running parallel scan requests
+ */
+export const PARALLEL_SCAN_CONCURRENCY_LIMIT = 10;
+
 /**
  * Max number of allowed attempts for batch write items
  */
diff --git a/packages/common/src/error/index.ts b/packages/common/src/error/index.ts
index 1340f92..c1867d2 100644
--- a/packages/common/src/error/index.ts
+++ b/packages/common/src/error/index.ts
@@ -18,3 +18,4 @@ export * from './missing-required-entity-physical-name-error';
 export * from './duplicate-entity-physical-name-error';
 export * from './invalid-unique-attribute-update-error';
 export * from './invalid-primary-key-attributes-update-error';
+export * from './invalid-parallel-scan-limit-option-error';
diff --git a/packages/common/src/error/invalid-parallel-scan-limit-option-error.ts b/packages/common/src/error/invalid-parallel-scan-limit-option-error.ts
new file mode 100644
index 0000000..a040b6b
--- /dev/null
+++ b/packages/common/src/error/invalid-parallel-scan-limit-option-error.ts
@@ -0,0 +1,10 @@
+export class InvalidParallelScanLimitOptionError extends Error {
+  name = 'InvalidParallelScanLimitOptionError';
+  constructor(limit: number, limitPerSegment: number) {
+    super();
+    this.message = `Invalid scan option "limit", when using parallel scan, value for "limitPerSegment" cannot be greater than "limit". Consider using a scan operation instead.    
+    limit: ${limit}
+    limitPerSegment: ${limitPerSegment}
+    `;
+  }
+}
diff --git a/packages/common/src/logger/debug-logger.ts b/packages/common/src/logger/debug-logger.ts
index 0d0cd4d..bedd845 100644
--- a/packages/common/src/logger/debug-logger.ts
+++ b/packages/common/src/logger/debug-logger.ts
@@ -22,6 +22,11 @@ export enum TRANSFORM_TRANSACTION_TYPE {
   TRANSACTION_READ = 'TRANSACTION_READ',
 }
 
+export enum TRANSFORM_SCAN_TYPE {
+  SCAN = 'SCAN',
+  PARALLEL_SCAN = 'PARALLEL_SCAN',
+}
+
 export enum TRANSFORM_BATCH_TYPE {
   BATCH_WRITE = 'BATCH_WRITE',
   BATCH_READ = 'BATCH_READ',
@@ -173,10 +178,12 @@ export class DebugLogger {
   logTransformScan({
     requestId,
     prefix,
+    operation,
     body,
     options,
   }: {
     requestId?: string;
+    operation: TRANSFORM_SCAN_TYPE;
     prefix: string;
     body?: any;
     options?: any;
@@ -184,7 +191,7 @@ export class DebugLogger {
     if (this.debugTransformScanLog.enabled) {
       this.debugTransformScanLog(
         `${chalk.bold.bgCyanBright(requestId)} ${chalk.green(
-          'SCAN'
+          operation
         )} ${chalk.magenta(prefix)}:`,
         ...(body
           ? [
diff --git a/packages/core/src/classes/manager/scan-manager.ts b/packages/core/src/classes/manager/scan-manager.ts
index b239f64..549d1ca 100644
--- a/packages/core/src/classes/manager/scan-manager.ts
+++ b/packages/core/src/classes/manager/scan-manager.ts
@@ -1,10 +1,13 @@
 import {
   EntityTarget,
   INTERNAL_ENTITY_ATTRIBUTE,
+  InvalidParallelScanLimitOptionError,
   MANAGER_NAME,
+  PARALLEL_SCAN_CONCURRENCY_LIMIT,
   STATS_TYPE,
 } from '@typedorm/common';
 import {DynamoDB} from 'aws-sdk';
+import pLimit from 'p-limit';
 import {getUniqueRequestId} from '../../helpers/get-unique-request-id';
 import {Connection} from '../connection/connection';
 import {ProjectionKeys} from '../expression/expression-input-parser';
@@ -46,34 +49,143 @@ interface ScanManageBaseOptions<Entity, PartitionKey> {
   select?: ProjectionKeys<Entity>;
 }
 
-export type ScanManagerScanOptions = ScanManageBaseOptions<any, any> & {
+export interface ScanManagerFindOptions<Entity>
+  extends ScanManageBaseOptions<
+    Entity,
+    {} // empty object since all attributes including ones used in primary key can be used with filter
+  > {
   /**
-   * Entity to scan
-   * When one is provided, filter expression in auto updated to include filer condition for this entity
+   * Total number of segments to divide this scan in.
+   *
+   * @default none - all items are scanned sequentially
+   */
+  totalSegments?: number;
+
+  /**
+   * Limit to apply per segment.
+   *
+   * when no `totalSegments` is provided, this option is ignored
+   * @default none - limit to apply per segment
+   */
+  limitPerSegment?: number;
+
+  /**
+   * Item cursor, used for paginating a scan.
+   *
+   * When the `totalSegments` option is provided, this option should be of type {[segmentNo]: [Key]}
    * @default none
    */
-  entity?: EntityTarget<any>;
-};
-export type ScanManagerFindOptions<Entity> = ScanManageBaseOptions<
-  Entity,
-  // empty object since all attributes including ones used in primary key can be used with filter
-  {}
->;
+  cursor?:
+    | Record<number, ScanManageBaseOptions<Entity, {}>['cursor']>
+    | ScanManageBaseOptions<Entity, {}>['cursor'];
+
+  /**
+   * Max number of requests to run in parallel
+   *
+   * When requesting parallel scan on x segments, request are executed in parallel using Promise.all
+   * While it is okay to run small number of requests in parallel, it is often a good idea to enforce a concurrency controller to stop node from eating up the all the memory
+   *
+   * This parameter does exactly that. i.e if requested to run scan with `20,000` segments, and `requestsConcurrencyLimit` is set to `100`
+   * TypeDORM will make sure that there are always only `100` requests are running in parallel at any time until all `20,000` segments have finished processing.
+   *
+   * @default PARALLEL_SCAN_CONCURRENCY_LIMIT
+   */
+  requestsConcurrencyLimit?: number;
+}
 
 export type ScanManagerCountOptions<Entity> = Pick<
   ScanManageBaseOptions<
     Entity,
-    // empty object since all attributes including ones used in primary key can be used with filter
-    {}
+    {} // empty object since all attributes including ones used in primary key can be used with filter
   >,
   'scanIndex' | 'where'
 >;
 
+export interface ScanManagerParallelScanOptions
+  extends ScanManageBaseOptions<any, any> {
+  /**
+   * Total number of segments to divide this scan in
+   */
+  totalSegments: number;
+
+  /**
+   * Limit to apply per segment
+   *
+   * @default none - limit to apply per segment
+   */
+  limitPerSegment?: number;
+
+  /**
+   * Entity to run scan for.
+   *
+   * When one is provided, filter expression in auto updated to include a default filer condition for matching entity
+   *
+   * @default none
+   */
+  entity?: EntityTarget<any>;
+
+  /**
+   * Per segment cursor, where key is the segment number, and value is the cursor options for that segment
+   *
+   * @default none
+   */
+  cursor?: Record<number, ScanManagerScanOptions['cursor']>;
+
+  /**
+   * Max number of requests to run in parallel
+   *
+   * When requesting parallel scan on x segments, request are executed in parallel using Promise.all
+   * While it is okay to run small number of requests in parallel, it is often a good idea to enforce a concurrency controller to stop node from eating up the all the memory
+   *
+   * This parameter does exactly that. i.e if requested to run scan with `20,000` segments, and `requestsConcurrencyLimit` is set to `100`
+   * TypeDORM will make sure that there are always only `100` requests are running in parallel at any time until all `20,000` segments have finished processing.
+   *
+   * @default PARALLEL_SCAN_CONCURRENCY_LIMIT
+   */
+  requestsConcurrencyLimit?: number;
+}
+
+export interface ScanManagerScanOptions
+  extends ScanManageBaseOptions<any, any> {
+  /**
+   * Entity to scan
+   *
+   * When one is provided, filter expression in auto updated to include filer condition for this entity
+   *
+   * @default none
+   */
+  entity?: EntityTarget<any>;
+
+  /**
+   * Number of current segment
+   *
+   * @default none - scan is not segmented
+   */
+  segment?: number;
+
+  /**
+   * Total number of segments to divide this scan in
+   *
+   * @default none - all items are scanned sequentially
+   */
+  totalSegments?: number;
+
+  /**
+   * Limit to apply per segment
+   *
+   * @default none - limit to apply per segment
+   */
+  limitPerSegment?: number;
+}
+
 export class ScanManager {
+  private itemsFetchedSoFarTotalParallelCount: number;
+  private limit = pLimit(PARALLEL_SCAN_CONCURRENCY_LIMIT);
   private _dcScanTransformer: DocumentClientScanTransformer;
 
   constructor(private connection: Connection) {
     this._dcScanTransformer = new DocumentClientScanTransformer(connection);
+    this.itemsFetchedSoFarTotalParallelCount = 0;
   }
 
   /**
@@ -86,19 +198,35 @@ export class ScanManager {
     entityClass: EntityTarget<Entity>,
     findOptions?: ScanManagerFindOptions<Entity>,
     metadataOptions?: MetadataOptions
-  ): Promise<{
-    items: Entity[] | undefined;
-    cursor: DynamoDB.DocumentClient.Key | undefined;
-  }> {
+  ) {
     const requestId = getUniqueRequestId(metadataOptions?.requestId);
 
-    const response = await this.scan<Entity>(
-      {...findOptions, entity: entityClass} as ScanManagerScanOptions,
-      {
-        requestId,
-        returnConsumedCapacity: metadataOptions?.returnConsumedCapacity,
-      }
-    );
+    let response: {
+      items?: Entity[];
+      unknownItems?: DynamoDB.DocumentClient.AttributeMap[];
+      cursor?:
+        | DynamoDB.DocumentClient.Key
+        | Record<number, DynamoDB.DocumentClient.Key>;
+    };
+
+    if (findOptions?.totalSegments) {
+      (response = await this.parallelScan<Entity>({
+        ...findOptions,
+        entity: entityClass,
+      } as ScanManagerParallelScanOptions)),
+        {
+          requestId,
+          returnConsumedCapacity: metadataOptions?.returnConsumedCapacity,
+        };
+    } else {
+      response = await this.scan<Entity>(
+        {...findOptions, entity: entityClass} as ScanManagerScanOptions,
+        {
+          requestId,
+          returnConsumedCapacity: metadataOptions?.returnConsumedCapacity,
+        }
+      );
+    }
 
     if (response.unknownItems?.length) {
       // log warning for items that were skipped form the response
@@ -149,8 +277,126 @@ export class ScanManager {
   }
 
   /**
-   * Low level scan operation
-   * Perhaps you are looking for higher level ScanManager.find operation
+   * Scans all items from dynamo table in parallel while also respecting the max provisioned concurrency
+   * @param scanOptions Options for parallel scan
+   * @param metadataOptions Additional metadata options
+   */
+  async parallelScan<Entity>(
+    scanOptions: ScanManagerParallelScanOptions,
+    metadataOptions?: MetadataOptions
+  ): Promise<{
+    items: Entity[] | undefined;
+    unknownItems: DynamoDB.DocumentClient.AttributeMap[] | undefined;
+    cursor: Record<number, DynamoDB.DocumentClient.Key | undefined>;
+  }> {
+    // start with 0
+    this.itemsFetchedSoFarTotalParallelCount = 0;
+
+    const concurrencyLimit =
+      PARALLEL_SCAN_CONCURRENCY_LIMIT || scanOptions.requestsConcurrencyLimit;
+    const requestId = getUniqueRequestId(metadataOptions?.requestId);
+
+    if (scanOptions.requestsConcurrencyLimit) {
+      this.limit = pLimit(scanOptions.requestsConcurrencyLimit);
+    }
+
+    const parallelScanOptions: ScanManagerScanOptions[] = [];
+
+    if (
+      scanOptions?.limit &&
+      scanOptions?.limitPerSegment &&
+      scanOptions?.limit < scanOptions?.limitPerSegment
+    ) {
+      throw new InvalidParallelScanLimitOptionError(
+        scanOptions?.limit,
+        scanOptions?.limitPerSegment
+      );
+    }
+
+    for (let index = 0; index < scanOptions.totalSegments; index++) {
+      // only the cursor for same segment can be applied
+      const cursorForSegment = scanOptions.cursor
+        ? scanOptions.cursor[index]
+        : undefined;
+
+      parallelScanOptions.push({
+        ...scanOptions,
+        cursor: cursorForSegment,
+        segment: index,
+      });
+    }
+
+    this.connection.logger.logInfo({
+      requestId,
+      scope: MANAGER_NAME.SCAN_MANAGER,
+      log: `Running scan in parallel with ${scanOptions.totalSegments} segments.`,
+    });
+
+    if (concurrencyLimit < scanOptions.totalSegments) {
+      this.connection.logger.logInfo({
+        requestId,
+        scope: MANAGER_NAME.SCAN_MANAGER,
+        log: `Current request concurrency limit ${concurrencyLimit} is lower than requested segments count ${scanOptions.totalSegments}
+        So requests will be run in a batch of ${concurrencyLimit} at a time until all segments ${scanOptions.totalSegments} have processed.`,
+      });
+    }
+
+    const allPromisesResponse = await Promise.all(
+      parallelScanOptions.map(options =>
+        this.toLimited(this.scan<Entity>(options, metadataOptions))
+      )
+    );
+
+    // merge all responses
+    const response = allPromisesResponse.reduce(
+      (
+        acc: {
+          items: Entity[];
+          unknownItems: DynamoDB.DocumentClient.AttributeMap[];
+          cursor: Record<number, DynamoDB.DocumentClient.Key>;
+        },
+        current,
+        index
+      ) => {
+        if (current.items?.length) {
+          if (!acc.items) {
+            acc.items = [];
+          }
+          acc.items = [...acc.items, ...current.items];
+        }
+
+        if (current.unknownItems?.length) {
+          if (!acc.unknownItems) {
+            acc.unknownItems = [];
+          }
+          acc.unknownItems = [...acc.unknownItems, ...current.unknownItems];
+        }
+
+        if (current.cursor) {
+          if (!acc.cursor) {
+            acc.cursor = {};
+          }
+          acc.cursor = {
+            ...acc.cursor,
+            [index]: current.cursor,
+          };
+        }
+        return acc;
+      },
+      {} as {
+        items: Entity[];
+        unknownItems: DynamoDB.DocumentClient.AttributeMap[];
+        cursor: Record<number, DynamoDB.DocumentClient.Key>;
+      }
+    );
+
+    return response;
+  }
+
+  /**
+   * Low level scan operation.
+   *
+   * Perhaps you are looking for higher level ScanManager.find or ScanManager.parallelScan operation
    * @param scanOptions scan options to run scan with
    * @param metadataOptions any other metadata options
    */
@@ -165,7 +411,13 @@ export class ScanManager {
     const requestId = getUniqueRequestId(metadataOptions?.requestId);
 
     const dynamoScanInput = this._dcScanTransformer.toDynamoScanItem(
-      scanOptions,
+      {
+        ...scanOptions,
+        // if requested segmented scan, then apply segment limit or default to limit operator
+        limit: scanOptions?.totalSegments
+          ? scanOptions?.limitPerSegment
+          : scanOptions?.limit,
+      },
       {
         requestId,
         returnConsumedCapacity: metadataOptions?.returnConsumedCapacity,
@@ -225,6 +477,14 @@ export class ScanManager {
     items: DynamoDB.DocumentClient.ItemList;
     cursor?: DynamoDB.DocumentClient.Key;
   }> {
+    // return if the count is already met
+    if (limit && this.itemsFetchedSoFarTotalParallelCount >= limit) {
+      return {
+        items: itemsFetched,
+        cursor,
+      };
+    }
+
     const {
       LastEvaluatedKey,
       Items = [],
@@ -243,18 +503,26 @@ export class ScanManager {
       });
     }
 
+    // recheck if requested items limit is already met, may be other worker
+    // if so drop the result of current request and return
+    if (limit && this.itemsFetchedSoFarTotalParallelCount >= limit) {
+      return {
+        items: itemsFetched,
+        cursor,
+      };
+    }
+
     itemsFetched = [...itemsFetched, ...Items];
+    this.itemsFetchedSoFarTotalParallelCount += Items.length;
 
     if (LastEvaluatedKey) {
-      if (!limit || itemsFetched.length < limit) {
-        return this._internalRecursiveScan({
-          scanInput,
-          limit,
-          cursor: LastEvaluatedKey,
-          itemsFetched,
-          metadataOptions,
-        });
-      }
+      return this._internalRecursiveScan({
+        scanInput,
+        limit,
+        cursor: LastEvaluatedKey,
+        itemsFetched,
+        metadataOptions,
+      });
     }
 
     return {
@@ -308,4 +576,13 @@ export class ScanManager {
 
     return currentCount;
   }
+
+  /**
+   * Simple wrapper to limit number of concurrent calls
+   * @param promise wraps promise in a limited factory
+   * @returns
+   */
+  private toLimited<T>(promise: Promise<T>) {
+    return this.limit(() => promise);
+  }
 }
diff --git a/packages/core/src/classes/transformer/document-client-scan-transformer.ts b/packages/core/src/classes/transformer/document-client-scan-transformer.ts
index 1c784ae..2de861d 100644
--- a/packages/core/src/classes/transformer/document-client-scan-transformer.ts
+++ b/packages/core/src/classes/transformer/document-client-scan-transformer.ts
@@ -6,6 +6,7 @@ import {
   NoSuchEntityExistsError,
   NoSuchIndexFoundError,
   QUERY_SELECT_TYPE,
+  TRANSFORM_SCAN_TYPE,
 } from '@typedorm/common';
 import {DynamoDB} from 'aws-sdk';
 import {isEmptyObject} from '../../helpers/is-empty-object';
@@ -23,6 +24,8 @@ interface ScanTransformerToDynamoScanOptions {
   where?: any;
   select?: any[];
   onlyCount?: boolean;
+  segment?: number; // current segment
+  totalSegments?: number; // total segments that this parallel scan will be divided in
 }
 
 export class DocumentClientScanTransformer extends LowOrderTransformers {
@@ -39,6 +42,7 @@ export class DocumentClientScanTransformer extends LowOrderTransformers {
   ): DynamoDB.DocumentClient.ScanInput {
     this.connection.logger.logTransformScan({
       requestId: metadataOptions?.requestId,
+      operation: TRANSFORM_SCAN_TYPE.SCAN,
       prefix: 'Before',
       options: scanOptions,
     });
@@ -169,9 +173,32 @@ export class DocumentClientScanTransformer extends LowOrderTransformers {
       }
     }
 
+    // validate value for segment and totalSegment before appending
+    if (
+      scanOptions?.totalSegments !== undefined &&
+      scanOptions?.totalSegments !== null
+    ) {
+      if (scanOptions?.totalSegments === 0) {
+        throw new Error(`Invalid scan option totalSegment: ${scanOptions?.totalSegments}.
+        totalSegments is optional, but when provided it's value must be greater than 0.`);
+      }
+      if (scanOptions?.segment === undefined || scanOptions?.segment === null) {
+        throw new Error(`Invalid scan option segment: ${scanOptions?.segment}.
+        When totalSegments value is defined, value for option 'segment' must also be defined.`);
+      }
+      if (scanOptions?.segment >= scanOptions?.totalSegments) {
+        throw new Error(`Invalid scan option segment: ${scanOptions?.segment}.
+        When totalSegments value is defined, value for option 'segment' must be one less than totalSegment size.`);
+      }
+
+      transformedScanInput.TotalSegments = scanOptions?.totalSegments;
+      transformedScanInput.Segment = scanOptions?.segment;
+    }
+
     this.connection.logger.logTransformScan({
       requestId: metadataOptions?.requestId,
       prefix: 'After',
+      operation: TRANSFORM_SCAN_TYPE.SCAN,
       body: transformedScanInput,
     });
 


Test patch:
diff --git a/packages/core/src/classes/manager/__test__/scan-manager.spec.ts b/packages/core/src/classes/manager/__test__/scan-manager.spec.ts
index 44d530d..4e85f90 100644
--- a/packages/core/src/classes/manager/__test__/scan-manager.spec.ts
+++ b/packages/core/src/classes/manager/__test__/scan-manager.spec.ts
@@ -1,3 +1,4 @@
+import {InvalidParallelScanLimitOptionError} from '@typedorm/common';
 import {table} from '@typedorm/core/__mocks__/table';
 import {User} from '@typedorm/core/__mocks__/user';
 import {UserAutoGenerateAttributes} from '@typedorm/core/__mocks__/user-auto-generate-attributes';
@@ -33,7 +34,7 @@ test('scan manager works', () => {
 /**
  * @group scan
  */
-test('scans table with given options', async () => {
+test('scans table with simple options', async () => {
   dcMock.scan.mockReturnValue({
     promise: () => ({
       Items: [
@@ -77,6 +78,54 @@ test('scans table with given options', async () => {
   );
 });
 
+test('scans table with parallel scan options', async () => {
+  dcMock.scan.mockReturnValue({
+    promise: () => ({
+      Items: [
+        {
+          id: '1',
+          PK: 'USER#1',
+          __en: 'user',
+        },
+        {
+          id: '2',
+          PK: 'USER#1',
+          __en: 'user',
+        },
+      ],
+    }),
+  });
+  const response = await manager.scan({
+    limitPerSegment: 1000,
+    segment: 0,
+    totalSegments: 2,
+    limit: 2000,
+  });
+
+  expect(dcMock.scan).toHaveBeenCalledTimes(1);
+  expect(dcMock.scan).toHaveBeenCalledWith({
+    Limit: 1000,
+    TableName: 'test-table',
+    Segment: 0,
+    TotalSegments: 2,
+  });
+  expect(response).toEqual({
+    items: [
+      {
+        id: '1',
+      },
+      {
+        id: '2',
+      },
+    ],
+  });
+  expect(
+    response.items?.forEach(item => {
+      expect(item).toBeInstanceOf(User);
+    })
+  );
+});
+
 test('scans table with multiple requests', async () => {
   dcMock.scan
     .mockImplementationOnce(() => ({
@@ -184,6 +233,92 @@ test('scans table with given options and returns deserialized and unknown items'
   });
 });
 
+/**
+ * @group parallelScan
+ */
+test('runs parallel scan requests for each segment', async () => {
+  const callIndex = {
+    0: 0,
+    1: 1,
+  };
+  dcMock.scan.mockImplementation(({Segment}: {Segment: 0 | 1}) => {
+    callIndex[Segment] = callIndex[Segment] + 1;
+    return {
+      promise: () => ({
+        Items: new Array(10).fill({
+          id: Segment,
+          __en: 'user',
+        }),
+        LastEvaluatedKey: {
+          PK: callIndex[Segment],
+        },
+      }),
+    };
+  });
+
+  const response = await manager.parallelScan({
+    totalSegments: 2,
+    limitPerSegment: 10,
+    limit: 20,
+  });
+
+  // Here, we are still making extra one request even tho we managed to get 20 items from the
+  // first requests on each segment, this is not 100% ideal but the only way possible to handle
+  // parallel requests, as we do not upfront know, what request will have how many items.
+  // However, we did manage to kill of further request from second segment
+  expect(dcMock.scan).toHaveBeenCalledTimes(3);
+  expect(response.cursor).toEqual({
+    '0': {
+      PK: 1,
+    },
+    '1': {
+      PK: 2,
+    },
+  });
+  expect(response.items?.length).toEqual(20);
+  expect(dcMock.scan.mock.calls).toEqual([
+    [
+      {
+        Limit: 10,
+        Segment: 0,
+        TableName: 'test-table',
+        TotalSegments: 2,
+      },
+    ],
+    [
+      {
+        Limit: 10,
+        Segment: 1,
+        TableName: 'test-table',
+        TotalSegments: 2,
+      },
+    ],
+    [
+      {
+        // request was rerun for that segment with pk
+        ExclusiveStartKey: {PK: 1},
+        Limit: 10,
+        Segment: 0,
+        TableName: 'test-table',
+        TotalSegments: 2,
+      },
+    ],
+  ]);
+});
+
+test('throws when trying to run parallel scan with invalid limit values', async () => {
+  const responseFac = () =>
+    manager.parallelScan({
+      totalSegments: 2,
+      limit: 100,
+      limitPerSegment: 200,
+    });
+
+  await expect(responseFac).rejects.toThrow(
+    InvalidParallelScanLimitOptionError
+  );
+});
+
 /**
  * @group find
  */
@@ -279,6 +414,67 @@ test('finds items from table with over the scan api with given options', async (
   });
 });
 
+test('finds items from table with using parallel scan', async () => {
+  dcMock.scan.mockImplementation(({Segment}) => ({
+    promise: () => ({
+      Items: [
+        {
+          id: Segment,
+          PK: `USER#${Segment}`,
+          __en: 'user',
+        },
+      ],
+    }),
+  }));
+  const response = await manager.find(User, {
+    totalSegments: 2,
+    where: {
+      id: {
+        GE: '1',
+      },
+    },
+  });
+
+  expect(dcMock.scan).toHaveBeenCalledTimes(2);
+  expect(dcMock.scan.mock.calls).toEqual([
+    [
+      {
+        ExpressionAttributeNames: {
+          '#FE___en': '__en',
+          '#FE_id': 'id',
+        },
+        ExpressionAttributeValues: {
+          ':FE___en': 'user',
+          ':FE_id': '1',
+        },
+        FilterExpression: '(#FE___en = :FE___en) AND (#FE_id >= :FE_id)',
+        Segment: 0,
+        TableName: 'test-table',
+        TotalSegments: 2,
+      },
+    ],
+    [
+      {
+        ExpressionAttributeNames: {
+          '#FE___en': '__en',
+          '#FE_id': 'id',
+        },
+        ExpressionAttributeValues: {
+          ':FE___en': 'user',
+          ':FE_id': '1',
+        },
+        FilterExpression: '(#FE___en = :FE___en) AND (#FE_id >= :FE_id)',
+        Segment: 1,
+        TableName: 'test-table',
+        TotalSegments: 2,
+      },
+    ],
+  ]);
+  expect(response).toEqual({
+    items: [{id: 0}, {id: 1}],
+  });
+});
+
 /**
  * @group count
  */
diff --git a/packages/core/src/classes/transformer/__test__/dc-scan-transformer.spec.ts b/packages/core/src/classes/transformer/__test__/dc-scan-transformer.spec.ts
index fe8c375..144a159 100644
--- a/packages/core/src/classes/transformer/__test__/dc-scan-transformer.spec.ts
+++ b/packages/core/src/classes/transformer/__test__/dc-scan-transformer.spec.ts
@@ -33,6 +33,9 @@ test('correctly extends low order transformers', () => {
   expect(dcScanTransformer.connection).toEqual(connection);
 });
 
+/**
+ * @group toDynamoScanItem
+ */
 test('transforms simple scan input', () => {
   const transformed = dcScanTransformer.toDynamoScanItem();
   expect(transformed).toEqual({TableName: 'test-table'});
@@ -127,6 +130,36 @@ test('transforms input with entity filter and option filter', () => {
   });
 });
 
+test('transforms simple parallel scan item', () => {
+  const transformed = dcScanTransformer.toDynamoScanItem({
+    totalSegments: 3,
+    segment: 0,
+  });
+
+  expect(transformed).toEqual({
+    Segment: 0,
+    TableName: 'test-table',
+    TotalSegments: 3,
+  });
+});
+
+test('throws when invalid values are configured for segments', () => {
+  const transformedFac = () =>
+    dcScanTransformer.toDynamoScanItem({
+      // total segment is 2 but current segment id is 3,
+      totalSegments: 2,
+      segment: 3,
+    });
+
+  expect(transformedFac).toThrow(
+    `Invalid scan option segment: 3.
+        When totalSegments value is defined, value for option 'segment' must be one less than totalSegment size.`
+  );
+});
+
+/**
+ * @group fromDynamoScanResponseItemList
+ */
 test('transforms simple dynamodb output items', () => {
   const transformed = dcScanTransformer.fromDynamoScanResponseItemList([
     {id: '1', __en: 'user', name: 'test-user'},