backfillNext method

Future<DVBackfillChunk?> backfillNext()

Copies the next chunk, or returns null when no rows remain.

Implementation

Future<DVBackfillChunk?> backfillNext() async {
  await _prepare();
  final List<Map<String, Object?>> last = await database.query(
    'SELECT chunk, last_key FROM $_chunks WHERE backfill = ? '
    'ORDER BY chunk DESC LIMIT 1',
    <Object?>[id],
  );
  final List<Map<String, Object?>> rows = last.isEmpty
      ? await database.query(
          'SELECT $key, $source FROM $table ORDER BY $key LIMIT ?',
          <Object?>[chunkSize],
        )
      : await database.query(
          'SELECT $key, $source FROM $table WHERE $key > ? '
          'ORDER BY $key LIMIT ?',
          <Object?>[_decodeKey(last.single['last_key']), chunkSize],
        );
  if (rows.isEmpty) {
    await _setState('complete', true);
    return null;
  }
  await _setState('complete', false);

  for (final Map<String, Object?> row in rows) {
    await database.execute(
      'UPDATE $table SET $target = ? WHERE $key = ?',
      <Object?>[convert(row[source]), row[key]],
    );
  }
  // Recorded after the rows are written. A crash between the two copies the
  // chunk again on restart, which is harmless; recording first would skip
  // rows that were never copied.
  final DVBackfillChunk chunk = DVBackfillChunk(
    index: last.isEmpty ? 0 : (last.single['chunk']! as num).toInt() + 1,
    first: rows.first[key],
    last: rows.last[key],
    rows: rows.length,
    state: 'backfilled',
    everMatched: false,
  );
  await database.execute(
    'INSERT INTO $_chunks (backfill, chunk, first_key, last_key, row_count, '
    'state, ever_matched) VALUES (?, ?, ?, ?, ?, ?, ?)',
    <Object?>[
      id,
      chunk.index,
      jsonEncode(chunk.first),
      jsonEncode(chunk.last),
      chunk.rows,
      chunk.state,
      0,
    ],
  );
  return chunk;
}