def _check_db_not_created(self, tablet):
   # Broadly catch all exceptions, since the exception being raised
   # is internal to MySQL.  We're strictly checking the error message
   # though, so should be fine.
   with self.assertRaisesRegexp(
       Exception, '(1049, "Unknown database \'%s\'")' % db_name):
     tablet.mquery(db_name, 'show tables')

  def _insert_values(self, tablet, id_offset, msg, keyspace_id, num_values):
    """Inserts values into MySQL along with the required routing comments.

    Args:
      tablet: the Tablet instance to modify.
      id: the value of `id` column.
      msg: the value of `msg` column.
      keyspace_id: the value of `keyspace_id` column.
    """

    # For maximum performance, multiple values are inserted in one statement.
    # However, when the statements are too long, queries will timeout and
    # vttablet will kill them. Therefore, we chunk it into multiple statements.
    def chunks(full_list, n):
      """Yield successive n-sized chunks from full_list."""
      for i in xrange(0, len(full_list), n):
        yield full_list[i:i+n]

    max_chunk_size = 100*1000
    k = utils.uint64_to_hex(keyspace_id)
    for chunk in chunks(range(1, num_values+1), max_chunk_size):
      logging.debug('Inserting values for range [%d, %d].', chunk[0], chunk[-1])
      values_str = ''
      for i in chunk:
        if i != chunk[0]:
          values_str += ','
        values_str += "(%d, '%s', 0x%x)" % (id_offset + i, msg, keyspace_id)
      tablet.mquery(
          'vt_test_keyspace', [
              'begin',
              'insert into worker_test(id, msg, keyspace_id) values%s '
              '/* vtgate:: keyspace_id:%s */' % (values_str, k),
              'commit'],
          write=True)

  def setUp(self):
    create_table = (
        'create table test (pk1 bigint, pk2 bigint, pk3 bigint, '
        'keyspace_id bigint, msg varchar(64), primary key (pk1, pk2, pk3)) '
        'Engine=InnoDB')
    destination_tablet.create_db('test_checkers')
    destination_tablet.mquery('test_checkers', create_table, True)
    for i, t in enumerate(source_tablets):
      t.create_db('test_checkers%s' % i)
      t.mquery('test_checkers%s' % i, create_table, True)

    destination_queries = []
    source_queries = [[] for t in source_tablets]
    for i in range(1, 400):
      query = (
          'insert into test (pk1, pk2, pk3, msg, keyspace_id) '
          "values (%s, %s, %s, 'message %s', %s)" % (i/100+1, i/10+1, i, i, i))
      destination_queries.append(query)
      source_queries[i % 2].append(query)
    for i in range(1100, 1110):
      query = (
          'insert into test (pk1, pk2, pk3, msg, keyspace_id) '
          "values (%s, %s, %s, 'message %s', %s)" % (i/100+1, i/10+1, i, i, i))
      source_queries[0].append(query)

    destination_tablet.mquery('test_checkers', destination_queries, write=True)
    for i, (tablet, queries) in enumerate(zip(source_tablets, source_queries)):
      tablet.mquery('test_checkers%s' % i, queries, write=True)
    self.c = self.make_checker()

 def _backfill_keyspace_id(self, tablet):
   tablet.mquery('vt_test_keyspace', [
       'begin',
       'update resharding1 set keyspace_id=0x1000000000000000 where id=1',
       'update resharding1 set keyspace_id=0x9000000000000000 where id=2',
       'update resharding1 set keyspace_id=0xD000000000000000 where id=3',
       'commit'
       ], write=True)

 def _insert_value(self, tablet, table, id, msg, keyspace_id):
   k = utils.uint64_to_hex(keyspace_id)
   tablet.mquery(
       'vt_test_keyspace',
       ['begin',
        'insert into %s(id, msg, keyspace_id) '
        'values(%d, "%s", 0x%x) /* vtgate:: keyspace_id:%s */ /* user_id:%d */' %
        (table, id, msg, keyspace_id, k, id),
        'commit'],
       write=True)

 def _insert_value(self, tablet, table, id, msg, keyspace_id):
   if keyspace_id_type == keyrange_constants.KIT_BYTES:
     k = base64.b64encode(pack_keyspace_id(keyspace_id))
   else:
     k = "%u" % keyspace_id
   tablet.mquery('vt_test_keyspace', [
       'begin',
       'insert into %s(id, msg, keyspace_id) values(%u, "%s", 0x%x) /* EMD keyspace_id:%s user_id:%u */' % (table, id, msg, keyspace_id, k, id),
       'commit'
       ], write=True)

 def _backfill_keyspace_id(self, tablet):
     tablet.mquery(
         "vt_test_keyspace",
         [
             "begin",
             "update resharding1 set keyspace_id=0x1000000000000000 where id=1",
             "update resharding1 set keyspace_id=0x9000000000000000 where id=2",
             "update resharding1 set keyspace_id=0xD000000000000000 where id=3",
             "commit",
         ],
         write=True,
     )

 def _insert_value(self, tablet, table, id, msg, keyspace_id):
     k = utils.uint64_to_hex(keyspace_id)
     tablet.mquery(
         "vt_test_keyspace",
         [
             "begin",
             "insert into %s(id, msg, keyspace_id) "
             'values(%d, "%s", 0x%x) /* vtgate:: keyspace_id:%s */ /* user_id:%d */'
             % (table, id, msg, keyspace_id, k, id),
             "commit",
         ],
         write=True,
     )

  def _insert_value(self, tablet, id, msg, keyspace_id):
    """Inserts a value in the MySQL database along with the required routing comments.

    Args:
      tablet - the Tablet instance to insert into
      id - the value of `id` column
      msg - the value of `msg` column
      keyspace_id - the value of `keyspace_id` column
    """
    k = "%u" % keyspace_id
    tablet.mquery('vt_test_keyspace', [
        'begin',
        'insert into worker_test(id, msg, keyspace_id) values(%u, "%s", 0x%x) /* EMD keyspace_id:%s user_id:%u */' % (id, msg, keyspace_id, k, id),
        'commit'
        ], write=True)

 def _check_tables(self, tablet, expectedCount):
     tables = tablet.mquery(db_name, "show tables")
     self.assertEqual(
         len(tables),
         expectedCount,
         "Unexpected table count on %s (not %d): got tables: %s" % (tablet.tablet_alias, expectedCount, str(tables)),
     )

 def _check_values(self, tablet, dbname, table, first, count):
     logging.debug("Checking %d values from %s/%s starting at %d", count, dbname, table, first)
     rows = tablet.mquery(dbname, "select id, msg from %s where id>=%d order by id limit %d" % (table, first, count))
     self.assertEqual(count, len(rows), "got wrong number of rows: %d != %d" % (len(rows), count))
     for i in xrange(count):
         self.assertEqual(first + i, rows[i][0], "invalid id[%d]: %d != %d" % (i, first + i, rows[i][0]))
         self.assertEqual(
             "value %d" % (first + i), rows[i][1], "invalid msg[%d]: 'value %d' != '%s'" % (i, first + i, rows[i][1])
         )

 def _check_vt_insert_test(self, tablet, index):
     # wait until it gets the data
     timeout = 10.0
     while True:
         result = tablet.mquery("vt_test_keyspace", "select msg from vt_insert_test where id=%d" % index)
         if len(result) == 1:
             break
         timeout = utils.wait_step(
             "waiting for replication to catch up on %s" % tablet.tablet_alias, timeout, sleep_time=0.1
         )

def check_value(tablet, table, id, msg, keyspace_id, should_be_here=True):
  result = tablet.mquery('vt_test_keyspace', 'select id, msg, keyspace_id from %s where id=%u' % (table, id))
  if should_be_here:
    if len(result) != 1:
      raise utils.TestError("Missing row in tablet %s for id=%u, keyspace_id=%x" % (tablet.tablet_alias, id, keyspace_id))
    result = result[0]
    if result[0] != id or result[1] != msg or result[2] != keyspace_id:
      raise utils.TestError("Row mismatch in tablet %s for id=%u, keyspace_id=%x: %s" % (tablet.tablet_alias, id, keyspace_id, str(result)))
  else:
    if len(result) != 0:
      raise utils.TestError("Extra row in tablet %s for id=%u, keyspace_id=%x: %s" % (tablet.tablet_alias, id, keyspace_id, str(result[0])))

  def _insert_values(self, tablet, id_offset, msg, keyspace_id, num_values):
    """Inserts values in the MySQL database along with the required routing comments.

    Args:
      tablet - the Tablet instance to insert into
      id - the value of `id` column
      msg - the value of `msg` column
      keyspace_id - the value of `keyspace_id` column
    """
    k = "%u" % keyspace_id
    values_str = ''
    for i in xrange(num_values):
      if i != 0:
        values_str += ','
      values_str += '(%u, "%s", 0x%x)' % (id_offset + i, msg, keyspace_id)
    tablet.mquery('vt_test_keyspace', [
        'begin',
        'insert into worker_test(id, msg, keyspace_id) values%s /* EMD keyspace_id:%s*/' % (values_str, k),
        'commit'
        ], write=True)

 def _check_values(self, tablet, dbname, table, first, count):
   logging.debug('Checking %u values from %s/%s starting at %u', count, dbname,
                 table, first)
   rows = tablet.mquery(dbname, 'select id, msg from %s where id>=%u order by id limit %u' % (table, first, count))
   self.assertEqual(count, len(rows), 'got wrong number of rows: %u != %u' %
                    (len(rows), count))
   for i in xrange(count):
     self.assertEqual(first + i, rows[i][0], 'invalid id[%u]: %u != %u' %
                      (i, first + i, rows[i][0]))
     self.assertEqual('value %u' % (first + i), rows[i][1],
                      "invalid msg[%u]: 'value %u' != '%s'" %
                      (i, first + i, rows[i][1]))

  def _insert_values(self, tablet, id_offset, msg, keyspace_id, num_values):
    """Inserts values into MySQL along with the required routing comments.

    Args:
      tablet: the Tablet instance to modify.
      id: the value of `id` column.
      msg: the value of `msg` column.
      keyspace_id: the value of `keyspace_id` column.
    """
    k = '%d' % keyspace_id
    values_str = ''
    for i in xrange(num_values):
      if i != 0:
        values_str += ','
      values_str += "(%d, '%s', 0x%x)" % (id_offset + i, msg, keyspace_id)
    tablet.mquery(
        'vt_test_keyspace', [
            'begin',
            'insert into worker_test(id, msg, keyspace_id) values%s '
            '/* EMD keyspace_id:%s*/' % (values_str, k),
            'commit'],
        write=True)

def direct_batch_write(count, tablet):
  """Writes a number of rows directly to MySQL on a tablet.

  This is significantly faster than do_writes(), but it works by bypassing the
  entire VtTablet layer, and batches all the inserts into a single round-trip.
  This can cause unexpected behavior, so should be used sparingly.
  """
  master_conn = get_connection(db_type='master')
  master_conn.begin()
  master_conn._execute('delete from vt_insert_test', {})
  master_conn.commit()
  kid_list = shard_kid_map[master_conn.shard]
  values_str = ''
  for x in xrange(count):
    if x != 0:
      values_str += ','
    keyspace_id = kid_list[count%len(kid_list)]
    values_str += '("test %s", "%s")' % (x, keyspace_id)
  tablet.mquery('vt_test_keyspace', [
        'begin',
        'insert into vt_insert_test(msg, keyspace_id) values%s' % values_str,
        'commit'
        ], write=True)

  def setUp(self):
    source_create_table = "create table test (pk1 bigint, k2 bigint, k3 bigint, keyspace_id bigint, msg varchar(64), primary key (pk1)) Engine=InnoDB"
    destination_create_table = "create table test_lookup (pk1_lookup bigint, msg_lookup varchar(64), primary key (pk1_lookup)) Engine=InnoDB"
    destination_tablet.create_db("test_checkers")
    destination_tablet.mquery("test_checkers", destination_create_table, True)

    for i, t in enumerate(source_tablets):
      t.create_db("test_checkers%s" % i)
      t.mquery("test_checkers%s" % i, source_create_table, True)

    destination_queries = []
    source_queries = [[] for t in source_tablets]
    for i in range(1, 400):
      destination_queries.append("insert into test_lookup (pk1_lookup, msg_lookup) values (%s, 'message %s')" % (i, i))
      source_queries[i % 2].append("insert into test (pk1, k2, k3, msg, keyspace_id) values (%s, %s, %s, 'message %s', %s)" % (i, i, i, i, i))
    for i in range(1100, 1110):
      query = "insert into test (pk1, k2, k3, msg, keyspace_id) values (%s, %s, %s, 'message %s', %s)" % (i, i, i, i, i)
      source_queries[0].append(query)

    destination_tablet.mquery("test_checkers", destination_queries, write=True)
    for i, (tablet, queries) in enumerate(zip(source_tablets, source_queries)):
      tablet.mquery("test_checkers%s" % i, queries, write=True)
    self.c = self.make_checker(destination_table_name="test_lookup", source_table_name="test", source_column_map={'pk1_lookup': 'pk1', 'msg_lookup': 'msg'})

 def _insert_value(self, tablet, table, id, msg, keyspace_id):
   tablet.mquery('vt_test_keyspace', [
       'begin',
       'insert into %s(id, msg, keyspace_id) values(%u, "%s", 0x%x) /* EMD keyspace_id:%u user_id:%u */' % (table, id, msg, keyspace_id, keyspace_id, id),
       'commit'
       ], write=True)

def check_tables(tablet, expectedCount):
    tables = tablet.mquery("vt_test_keyspace", "show tables")
    if len(tables) != expectedCount:
        raise utils.TestError(
            "Unexpected table count on %s (not %u): %s" % (tablet.tablet_alias, expectedCount, str(tables))
        )