chunkedProcess method

@override Future<void> chunkedProcess({ required int chunkSize, required Future<List<Map<String, dynamic>>> processor(List<Map<String, dynamic>> chunk), String? destination, List<String> columns = const ['*'], })

Implementation

@override
Future<void> chunkedProcess({
  required int chunkSize,
  required Future<List<Map<String, dynamic>>> Function(
          List<Map<String, dynamic>> chunk)
      processor,
  String? destination,
  List<String> columns = const ['*'],
}) async {
  int offset = 0;

  while (true) {
    final chunkQuery = QueryBuilderImpl()
      ..table(getTable)
      ..select(columns)
      ..limit(chunkSize)
      ..offset(offset);

    final chunk = await chunkQuery.get();

    if (chunk.isEmpty) break;

    final processedChunk = await processor(chunk);

    if (destination != null && processedChunk.isNotEmpty) {
      final insertQuery = QueryBuilderImpl()..table(destination);
      await insertQuery.insertMany(processedChunk);
    }

    if (chunk.length < chunkSize) break;

    offset += chunkSize;
  }
}