def _PerformTrickle(self, batch_size):
    starting_offset = self._offset_bytes
    self._fh.seek(self._offset_bytes)

    # pre-read the data to avoid interleaving disk reads with database writes.
    #
    # also, break the input file into batches.  if we see a BEGIN then
    # we must read through to the next COMMIT even if we would be
    # bigger than batch size.

    # batch is a list of lists.  each entry represents a transaction
    # (ie, list of statements) to be applied to the database.
    batch = []
    while len(batch) < batch_size:
      line = self._fh.readline()
      if not line:
        break

      if not ValidSqlLine(line):
        logging.fatal('Encountered invalid sql: %s', line)

      if begin_re.match(line):
        transaction = []
        while 1:
          next_line = self._fh.readline()

          if not next_line:
            logging.fatal('Input file terminated inside of BEGIN block')
          elif not ValidSqlLine(next_line):
            logging.fatal('Encountered invalid sql: %s', next_line)
          elif begin_re.match(next_line):
            logging.fatal('Attempt to nest transactions')
          elif commit_re.match(next_line):
            break
          else:
            transaction.append(next_line)
        batch.append(transaction)
      elif commit_re.match(line):
        logging.fatal('Attempt to commit outside of transaction')
      else:
        batch.append([line])

    self._offset_bytes = self._fh.tell()
    with self._db.Transaction():
      for transaction in batch:
        for statement in transaction:
          try:
            self._db.ExecuteOrDie(statement)
          except db.QueryWarningsException as e:
            logging.warn('SQL generated warning: %s, %s', str(e), statement)
            if not FLAGS.allow_warnings:
              raise

      result = self._db.ExecuteOrDie(
          'UPDATE %s.%s SET Offset = %d WHERE Filename = "%s" AND Offset = %d' %
          (self._state_database, self._state_table,
           self._offset_bytes, self._filename, starting_offset))
      if result.GetRowsAffected() != 1:
        logging.fatal('Attempt to update database state but something '
                      'already changed it')

    return len(batch)
Stores generic variables in the session prepending it with _GENERIC_VAR_KEY_PREFIX.
dataset/ETHPy150Open google/mysql-tools/trickle_file_runner.py/DbFileTrickler._PerformTrickle--crossover with--dataset/ETHPy150Open Tivix/django-common/django_common/session.py/SessionManager.generic_var