def setup_unsharded_keyspace():
    utils.run_vtctl(["CreateKeyspace", UNSHARDED_KEYSPACE])
    utils.run_vtctl(["SetKeyspaceShardingInfo", "-force", UNSHARDED_KEYSPACE, "keyspace_id", "uint64"])
    unsharded_master.init_tablet("master", keyspace=UNSHARDED_KEYSPACE, shard="0")
    unsharded_replica.init_tablet("replica", keyspace=UNSHARDED_KEYSPACE, shard="0")

    utils.run_vtctl(["RebuildKeyspaceGraph", UNSHARDED_KEYSPACE], auto_log=True)

    for t in [unsharded_master, unsharded_replica]:
        t.create_db("vt_test_keyspace_unsharded")
        t.mquery(unsharded_master.dbname, create_vt_insert_test)
        t.start_vttablet(wait_for_state=None)

    for t in [unsharded_master, unsharded_replica]:
        t.wait_for_vttablet_state("SERVING")

    utils.run_vtctl(["InitShardMaster", "%s/0" % UNSHARDED_KEYSPACE, unsharded_master.tablet_alias], auto_log=True)

    utils.run_vtctl(["RebuildKeyspaceGraph", UNSHARDED_KEYSPACE], auto_log=True)

    utils.check_srv_keyspace(
        "test_nj",
        UNSHARDED_KEYSPACE,
        "Partitions(master): -\n" + "Partitions(rdonly): -\n" + "Partitions(replica): -\n",
    )

  def verify(self):
    self.assert_shard_data_equal(0, worker.shard_master,
                                 worker.shard_0_tablets.replica)
    self.assert_shard_data_equal(1, worker.shard_master,
                                 worker.shard_1_tablets.replica)

    # Verify effect of MigrateServedTypes. Dest shards are serving now.
    utils.check_srv_keyspace('test_nj', self.KEYSPACE,
                             'Partitions(master): -80 80-\n'
                             'Partitions(rdonly): -80 80-\n'
                             'Partitions(replica): -80 80-\n')

    # source shard: query service must be disabled after MigrateServedTypes.
    source_shards = [worker.shard_rdonly1,
                     worker.shard_replica,
                     worker.shard_master]
    for shard in source_shards:
      utils.check_tablet_query_service(
          self, shard, serving=False, tablet_control_disabled=True)

    # dest shard -80, 80-: query service must be enabled
    # after MigrateServedTypes.
    dest_shards = [worker.shard_0_rdonly1,
                   worker.shard_0_replica,
                   worker.shard_0_master,
                   worker.shard_1_rdonly1,
                   worker.shard_1_replica,
                   worker.shard_1_master]
    for shard in dest_shards:
      utils.check_tablet_query_service(
          self,
          shard,
          serving=True,
          tablet_control_disabled=False)

def start_tablets():
  global __tablets
  # start tablets
  for t in __tablets:
    t.start_vttablet(wait_for_state=None)

  # wait for them to come in serving state
  for t in __tablets:
    t.wait_for_vttablet_state('SERVING')

  # InitShardMaster for master tablets
  for t in __tablets:
    if t.tablet_type == 'master':
      utils.run_vtctl(['InitShardMaster', t.keyspace+'/'+t.shard,
                       t.tablet_alias], auto_log=True)

  for ks in topo_schema.keyspaces:
    ks_name = ks[0]
    ks_type = ks[1]
    utils.run_vtctl(['RebuildKeyspaceGraph', ks_name],
                     auto_log=True)
    if ks_type == shard_constants.RANGE_SHARDED:
      utils.check_srv_keyspace('test_nj', ks_name,
                               'Partitions(master): -80 80-\n'
                               'Partitions(rdonly): -80 80-\n'
                               'Partitions(replica): -80 80-\n')

def setup_sharded_keyspace():
  utils.run_vtctl(['CreateKeyspace', SHARDED_KEYSPACE])
  utils.run_vtctl(['SetKeyspaceShardingInfo', '-force', SHARDED_KEYSPACE,
                   'keyspace_id', 'uint64'])
  shard_0_master.init_tablet('master', keyspace=SHARDED_KEYSPACE, shard='-80')
  shard_0_replica.init_tablet(
      'replica', keyspace=SHARDED_KEYSPACE, shard='-80')
  shard_1_master.init_tablet('master', keyspace=SHARDED_KEYSPACE, shard='80-')
  shard_1_replica.init_tablet(
      'replica', keyspace=SHARDED_KEYSPACE, shard='80-')

  utils.run_vtctl(['RebuildKeyspaceGraph', SHARDED_KEYSPACE,], auto_log=True)

  for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
    t.create_db('vt_test_keyspace_sharded')
    t.mquery(shard_0_master.dbname, create_vt_insert_test)
    t.start_vttablet(wait_for_state=None)

  for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
    t.wait_for_vttablet_state('SERVING')

  utils.run_vtctl(['InitShardMaster', '%s/-80' % SHARDED_KEYSPACE,
                   shard_0_master.tablet_alias], auto_log=True)
  utils.run_vtctl(['InitShardMaster', '%s/80-' % SHARDED_KEYSPACE,
                   shard_1_master.tablet_alias], auto_log=True)

  utils.run_vtctl(['RebuildKeyspaceGraph', SHARDED_KEYSPACE],
                  auto_log=True)

  utils.check_srv_keyspace('test_nj', SHARDED_KEYSPACE,
                           'Partitions(master): -80 80-\n'
                           'Partitions(rdonly): -80 80-\n'
                           'Partitions(replica): -80 80-\n')

def setup_unsharded_keyspace():
  utils.run_vtctl(['CreateKeyspace', UNSHARDED_KEYSPACE])
  utils.run_vtctl(['SetKeyspaceShardingInfo', '-force', UNSHARDED_KEYSPACE,
                   'keyspace_id', 'uint64'])
  unsharded_master.init_tablet('master', keyspace=UNSHARDED_KEYSPACE, shard='0')
  unsharded_replica.init_tablet('replica', keyspace=UNSHARDED_KEYSPACE, shard='0')
  unsharded_rdonly.init_tablet('rdonly', keyspace=UNSHARDED_KEYSPACE, shard='0')

  utils.run_vtctl(['RebuildKeyspaceGraph', UNSHARDED_KEYSPACE,], auto_log=True)

  for t in [unsharded_master, unsharded_replica, unsharded_rdonly]:
    t.create_db('vt_test_keyspace_unsharded')
    t.mquery(unsharded_master.dbname, create_vt_insert_test)
    t.start_vttablet(wait_for_state=None)

  for t in [unsharded_master, unsharded_replica, unsharded_rdonly]:
    t.wait_for_vttablet_state('SERVING')

  utils.run_vtctl(['ReparentShard', '-force', '%s/0' % UNSHARDED_KEYSPACE,
                   unsharded_master.tablet_alias], auto_log=True)

  utils.run_vtctl(['RebuildKeyspaceGraph', UNSHARDED_KEYSPACE],
                   auto_log=True)

  utils.check_srv_keyspace('test_nj', UNSHARDED_KEYSPACE,
                           'Partitions(master): -\n' +
                           'Partitions(rdonly): -\n' +
                           'Partitions(replica): -\n' +
                           'TabletTypes: master,rdonly,replica')

def setup_sharded_keyspace():
  utils.run_vtctl(['CreateKeyspace', SHARDED_KEYSPACE])
  utils.run_vtctl(['SetKeyspaceShardingInfo', '-force', SHARDED_KEYSPACE,
                   'keyspace_id', 'uint64'])
  shard_0_master.start_vttablet(
      wait_for_state=None, target_tablet_type='replica',
      init_keyspace=SHARDED_KEYSPACE, init_shard='-80')
  shard_0_replica.start_vttablet(
      wait_for_state=None, target_tablet_type='replica',
      init_keyspace=SHARDED_KEYSPACE, init_shard='-80')
  shard_1_master.start_vttablet(
      wait_for_state=None, target_tablet_type='replica',
      init_keyspace=SHARDED_KEYSPACE, init_shard='80-')
  shard_1_replica.start_vttablet(
      wait_for_state=None, target_tablet_type='replica',
      init_keyspace=SHARDED_KEYSPACE, init_shard='80-')

  for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
    t.wait_for_vttablet_state('NOT_SERVING')

  utils.run_vtctl(['InitShardMaster', '-force', '%s/-80' % SHARDED_KEYSPACE,
                   shard_0_master.tablet_alias], auto_log=True)
  utils.run_vtctl(['InitShardMaster', '-force', '%s/80-' % SHARDED_KEYSPACE,
                   shard_1_master.tablet_alias], auto_log=True)
  utils.run_vtctl(['ApplySchema', '-sql', create_vt_insert_test,
                   SHARDED_KEYSPACE])

  utils.check_srv_keyspace('test_nj', SHARDED_KEYSPACE,
                           'Partitions(master): -80 80-\n'
                           'Partitions(rdonly): -80 80-\n'
                           'Partitions(replica): -80 80-\n')

    def verify(self):
        self.assert_shard_data_equal(0, worker.shard_master, worker.shard_0_tablets.replica)
        self.assert_shard_data_equal(1, worker.shard_master, worker.shard_1_tablets.replica)

        # Verify effect of MigrateServedTypes. Dest shards are serving now.
        utils.check_srv_keyspace(
            "test_nj",
            self.KEYSPACE,
            "Partitions(master): -80 80-\n" "Partitions(rdonly): -80 80-\n" "Partitions(replica): -80 80-\n",
        )

        # Check that query service is disabled (source shard) or enabled (dest).

        # The 'rdonly' tablet requires an explicit healthcheck first because
        # the following sequence of events is happening in this test:
        # - SplitDiff returns 'rdonly' as 'spare' tablet (NOT_SERVING)
        # - MigrateServedTypes runs and does not refresh then 'spare' tablet
        #   (still NOT_SERVING)
        #   Shard_TabletControl.DisableQueryService=true will be set in the topology
        # - explicit or periodic healthcheck runs:
        #   a) tablet seen as caught up, change type from 'spare' to 'rdonly'
        #      (change to SERVING)
        #   b) post-action callback agent.refreshTablet() reads the topology
        #      and finds out that DisableQueryService=true is set.
        #      (change to NOT_SERVING)
        #
        # We must run an explicit healthcheck or we can see one of the two states:
        # - NOT_SERVING, DisableQueryService=false, tablet type 'spare'
        #   (immediately after SplitDiff returned)
        # - SERVING, DisableQueryService=false, tablet type 'rdonly'
        #   (during healthcheck before post-action callback is called)
        utils.run_vtctl(["RunHealthCheck", worker.shard_rdonly1.tablet_alias], auto_log=True)

        # source shard: query service must be disabled after MigrateServedTypes.
        utils.check_tablet_query_service(self, worker.shard_rdonly1, serving=False, tablet_control_disabled=True)
        utils.check_tablet_query_service(self, worker.shard_replica, serving=False, tablet_control_disabled=True)
        utils.check_tablet_query_service(self, worker.shard_master, serving=False, tablet_control_disabled=True)

        # dest shard -80: query service must be disabled after MigrateServedTypes.
        # Run explicit healthcheck because 'rdonly' tablet may still be 'spare'.
        utils.run_vtctl(["RunHealthCheck", worker.shard_0_rdonly1.tablet_alias], auto_log=True)
        utils.check_tablet_query_service(self, worker.shard_0_rdonly1, serving=True, tablet_control_disabled=False)
        utils.check_tablet_query_service(self, worker.shard_0_replica, serving=True, tablet_control_disabled=False)
        utils.check_tablet_query_service(self, worker.shard_0_master, serving=True, tablet_control_disabled=False)

        # dest shard 80-: query service must be disabled after MigrateServedTypes.
        # Run explicit healthcheck because 'rdonly' tablet is still 'spare'.
        utils.run_vtctl(["RunHealthCheck", worker.shard_1_rdonly1.tablet_alias], auto_log=True)
        utils.check_tablet_query_service(self, worker.shard_1_rdonly1, serving=True, tablet_control_disabled=False)
        utils.check_tablet_query_service(self, worker.shard_1_replica, serving=True, tablet_control_disabled=False)
        utils.check_tablet_query_service(self, worker.shard_1_master, serving=True, tablet_control_disabled=False)

def setup_tablets():
  # Start up a master mysql and vttablet
  logging.debug('Setting up tablets')
  utils.run_vtctl(['CreateKeyspace', KEYSPACE_NAME])
  utils.run_vtctl(['SetKeyspaceShardingInfo', '-force', KEYSPACE_NAME,
                   'keyspace_id', 'uint64'])
  shard_0_master.init_tablet(
      'master',
      keyspace=KEYSPACE_NAME,
      shard='0',
      tablet_index=0)
  shard_0_replica1.init_tablet(
      'replica',
      keyspace=KEYSPACE_NAME,
      shard='0',
      tablet_index=1)

  utils.run_vtctl(['RebuildKeyspaceGraph', KEYSPACE_NAME], auto_log=True)

  for t in [shard_0_master, shard_0_replica1]:
    t.create_db('vt_test_keyspace')
    for create_table in create_tables:
      t.mquery(shard_0_master.dbname, create_table)
    t.start_vttablet(wait_for_state=None, target_tablet_type='replica')

  for t in [shard_0_master]:
    t.wait_for_vttablet_state('SERVING')
  for t in [shard_0_replica1]:
    t.wait_for_vttablet_state('NOT_SERVING')

  utils.run_vtctl(['InitShardMaster', KEYSPACE_NAME+'/0',
                   shard_0_master.tablet_alias], auto_log=True)

  for t in [shard_0_replica1]:
    utils.wait_for_tablet_type(t.tablet_alias, 'replica')

  for t in [shard_0_master, shard_0_replica1]:
    t.wait_for_vttablet_state('SERVING')

  utils.run_vtctl(
      ['RebuildKeyspaceGraph', KEYSPACE_NAME], auto_log=True)

  utils.check_srv_keyspace(
      'test_nj', KEYSPACE_NAME,
      'Partitions(master): -\n'
      'Partitions(rdonly): -\n'
      'Partitions(replica): -\n')

def setup_tablets():
  global vtgate_server
  global vtgate_port

  # Start up a master mysql and vttablet
  logging.debug("Setting up tablets")
  utils.run_vtctl(['CreateKeyspace', KEYSPACE_NAME])
  utils.run_vtctl(['SetKeyspaceShardingInfo', '-force', KEYSPACE_NAME,
                   'keyspace_id', 'uint64'])
  shard_0_master.init_tablet('master', keyspace=KEYSPACE_NAME, shard='-80')
  shard_0_replica.init_tablet('replica', keyspace=KEYSPACE_NAME, shard='-80')
  shard_1_master.init_tablet('master', keyspace=KEYSPACE_NAME, shard='80-')
  shard_1_replica.init_tablet('replica', keyspace=KEYSPACE_NAME, shard='80-')

  utils.run_vtctl(['RebuildKeyspaceGraph', KEYSPACE_NAME], auto_log=True)

  for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
    t.create_db('vt_test_keyspace')
    for create_table in create_tables:
      t.mquery(shard_0_master.dbname, create_table)
    t.start_vttablet(wait_for_state=None)

  for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
    t.wait_for_vttablet_state('SERVING')

  utils.run_vtctl(['ReparentShard', '-force', KEYSPACE_NAME+'/-80',
                   shard_0_master.tablet_alias], auto_log=True)
  utils.run_vtctl(['ReparentShard', '-force', KEYSPACE_NAME+'/80-',
                   shard_1_master.tablet_alias], auto_log=True)

  utils.run_vtctl(['RebuildKeyspaceGraph', KEYSPACE_NAME],
                   auto_log=True)

  utils.check_srv_keyspace('test_nj', KEYSPACE_NAME,
                           'Partitions(master): -80 80-\n' +
                           'Partitions(replica): -80 80-\n' +
                           'TabletTypes: master,replica')

  vtgate_server, vtgate_port = utils.vtgate_start()
  # FIXME(shrutip): this should be removed once vtgate_cursor's
  # dependency on topology goes away.
  vtgate_client = zkocc.ZkOccConnection("localhost:%u" % vtgate_port,
                                        "test_nj", 30.0)
  topology.read_topology(vtgate_client)

def setup_tablets():
  global vtgate_server
  global vtgate_port

  # Start up a master mysql and vttablet
  logging.debug("Setting up tablets")
  utils.run_vtctl(['CreateKeyspace', 'test_keyspace'])
  utils.run_vtctl(['SetKeyspaceShardingInfo', '-force', 'test_keyspace',
                   'keyspace_id', 'uint64'])
  shard_0_master.init_tablet('master', keyspace='test_keyspace', shard='-80')
  shard_0_replica.init_tablet('replica', keyspace='test_keyspace', shard='-80')
  shard_1_master.init_tablet('master', keyspace='test_keyspace', shard='80-')
  shard_1_replica.init_tablet('replica', keyspace='test_keyspace', shard='80-')

  utils.run_vtctl(['RebuildKeyspaceGraph', 'test_keyspace'], auto_log=True)

  for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
    t.create_db('vt_test_keyspace')
    t.mquery(shard_0_master.dbname, create_vt_insert_test)
    t.mquery(shard_0_master.dbname, create_vt_a)
    t.start_vttablet(wait_for_state=None)

  for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
    t.wait_for_vttablet_state('SERVING')

  utils.run_vtctl(['ReparentShard', '-force', 'test_keyspace/-80',
                   shard_0_master.tablet_alias], auto_log=True)
  utils.run_vtctl(['ReparentShard', '-force', 'test_keyspace/80-',
                   shard_1_master.tablet_alias], auto_log=True)

  utils.run_vtctl(['RebuildKeyspaceGraph', 'test_keyspace'],
                   auto_log=True)

  utils.check_srv_keyspace('test_nj', 'test_keyspace',
                           'Partitions(master): -80 80-\n' +
                           'Partitions(replica): -80 80-\n' +
                           'TabletTypes: master,replica')

  vtgate_server, vtgate_port = utils.vtgate_start()
  vtgate_client = zkocc.ZkOccConnection("localhost:%u" % vtgate_port,
                                        "test_nj", 30.0)
  topology.read_topology(vtgate_client)

def setup_tablets():
  # Start up a master mysql and vttablet
  logging.debug('Setting up tablets')
  utils.run_vtctl(['CreateKeyspace', 'test_keyspace'])
  utils.run_vtctl(['SetKeyspaceShardingInfo', '-force', 'test_keyspace',
                   'keyspace_id', 'uint64'])
  shard_0_master.init_tablet('master', keyspace='test_keyspace', shard='-80')
  shard_0_replica.init_tablet('replica', keyspace='test_keyspace', shard='-80')
  shard_1_master.init_tablet('master', keyspace='test_keyspace', shard='80-')
  shard_1_replica.init_tablet('replica', keyspace='test_keyspace', shard='80-')

  utils.run_vtctl(['RebuildKeyspaceGraph', 'test_keyspace'], auto_log=True)

  for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
    t.create_db('vt_test_keyspace')
    t.mquery(shard_0_master.dbname, create_vt_insert_test)
    t.mquery(shard_0_master.dbname, create_vt_a)
    t.start_vttablet(wait_for_state=None)

  for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
    t.wait_for_vttablet_state('SERVING')

  utils.run_vtctl(
      ['InitShardMaster', 'test_keyspace/-80',
       shard_0_master.tablet_alias], auto_log=True)
  utils.run_vtctl(
      ['InitShardMaster', 'test_keyspace/80-',
       shard_1_master.tablet_alias], auto_log=True)

  utils.run_vtctl(
      ['RebuildKeyspaceGraph', 'test_keyspace'], auto_log=True)

  utils.check_srv_keyspace('test_nj', 'test_keyspace',
                           'Partitions(master): -80 80-\n'
                           'Partitions(rdonly): -80 80-\n'
                           'Partitions(replica): -80 80-\n')

  utils.VtGate().start()
  vtgate_client = zkocc.ZkOccConnection(utils.vtgate.addr(), 'test_nj', 30.0)
  topology.read_topology(vtgate_client)

def setup_unsharded_keyspace():
  utils.run_vtctl(['CreateKeyspace', UNSHARDED_KEYSPACE])
  utils.run_vtctl(['SetKeyspaceShardingInfo', '-force', UNSHARDED_KEYSPACE,
                   'keyspace_id', 'uint64'])

  unsharded_master.init_tablet(
      'master',
      keyspace=UNSHARDED_KEYSPACE,
      shard='0',
      tablet_index=0)
  unsharded_replica.init_tablet(
      'replica',
      keyspace=UNSHARDED_KEYSPACE,
      shard='0',
      tablet_index=1)

  for t in [unsharded_master, unsharded_replica]:
    t.create_db('vt_test_keyspace_unsharded')
    t.mquery(unsharded_master.dbname, create_vt_insert_test)
    t.start_vttablet(wait_for_state=None)

  for t in [unsharded_master]:
    t.wait_for_vttablet_state('SERVING')
  for t in [unsharded_replica]:
    t.wait_for_vttablet_state('NOT_SERVING')

  utils.run_vtctl(['InitShardMaster', '-force', '%s/0' % UNSHARDED_KEYSPACE,
                   unsharded_master.tablet_alias], auto_log=True)

  for t in [unsharded_replica]:
    utils.wait_for_tablet_type(t.tablet_alias, 'replica')
  for t in [unsharded_master, unsharded_replica]:
    t.wait_for_vttablet_state('SERVING')

  # rebuild to be sure we have the right version
  utils.run_vtctl(['RebuildKeyspaceGraph', UNSHARDED_KEYSPACE], auto_log=True)
  utils.check_srv_keyspace('test_nj', UNSHARDED_KEYSPACE,
                           'Partitions(master): -\n'
                           'Partitions(rdonly): -\n'
                           'Partitions(replica): -\n')

def setup_tablets():
  global vtgate_server
  global vtgate_port

  # Start up a master mysql and vttablet
  logging.debug("Setting up tablets")
  utils.run_vtctl(['CreateKeyspace', KEYSPACE_NAME])
  utils.run_vtctl(['SetKeyspaceShardingInfo', '-force', KEYSPACE_NAME,
                   'keyspace_id', 'uint64'])
  shard_0_master.init_tablet('master', keyspace=KEYSPACE_NAME, shard='-80')
  shard_0_replica.init_tablet('replica', keyspace=KEYSPACE_NAME, shard='-80')
  shard_1_master.init_tablet('master', keyspace=KEYSPACE_NAME, shard='80-')
  shard_1_replica.init_tablet('replica', keyspace=KEYSPACE_NAME, shard='80-')

  utils.run_vtctl(['RebuildKeyspaceGraph', KEYSPACE_NAME], auto_log=True)

  for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
    t.create_db('vt_test_keyspace')
    for create_table in create_tables:
      t.mquery(shard_0_master.dbname, create_table)
    t.start_vttablet(wait_for_state=None)

  for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
    t.wait_for_vttablet_state('SERVING')

  utils.run_vtctl(['ReparentShard', '-force', KEYSPACE_NAME+'/-80',
                   shard_0_master.tablet_alias], auto_log=True)
  utils.run_vtctl(['ReparentShard', '-force', KEYSPACE_NAME+'/80-',
                   shard_1_master.tablet_alias], auto_log=True)

  utils.run_vtctl(['RebuildKeyspaceGraph', KEYSPACE_NAME],
                   auto_log=True)

  utils.check_srv_keyspace('test_nj', KEYSPACE_NAME,
                           'Partitions(master): -80 80-\n' +
                           'Partitions(replica): -80 80-\n' +
                           'TabletTypes: master,replica')

  vtgate_server, vtgate_port = utils.vtgate_start()

def setup_tablets():
    global vtgate_server
    global vtgate_port

    # Start up a master mysql and vttablet
    logging.debug("Setting up tablets")
    utils.run_vtctl(["CreateKeyspace", "test_keyspace"])
    utils.run_vtctl(["SetKeyspaceShardingInfo", "-force", "test_keyspace", "keyspace_id", "uint64"])
    shard_0_master.init_tablet("master", keyspace="test_keyspace", shard="-80")
    shard_0_replica.init_tablet("replica", keyspace="test_keyspace", shard="-80")
    shard_1_master.init_tablet("master", keyspace="test_keyspace", shard="80-")
    shard_1_replica.init_tablet("replica", keyspace="test_keyspace", shard="80-")

    utils.run_vtctl(["RebuildKeyspaceGraph", "test_keyspace"], auto_log=True)

    for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
        t.create_db("vt_test_keyspace")
        t.mquery(shard_0_master.dbname, create_vt_insert_test)
        t.mquery(shard_0_master.dbname, create_vt_a)
        t.start_vttablet(wait_for_state=None)

    for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
        t.wait_for_vttablet_state("SERVING")

    utils.run_vtctl(["ReparentShard", "-force", "test_keyspace/-80", shard_0_master.tablet_alias], auto_log=True)
    utils.run_vtctl(["ReparentShard", "-force", "test_keyspace/80-", shard_1_master.tablet_alias], auto_log=True)

    utils.run_vtctl(["RebuildKeyspaceGraph", "test_keyspace"], auto_log=True)

    utils.check_srv_keyspace(
        "test_nj",
        "test_keyspace",
        "Partitions(master): -80 80-\n" + "Partitions(replica): -80 80-\n" + "TabletTypes: master,replica",
    )

    vtgate_server, vtgate_port = utils.vtgate_start()
    vtgate_client = zkocc.ZkOccConnection("localhost:%u" % vtgate_port, "test_nj", 30.0)
    topology.read_topology(vtgate_client)

def setup_tablets():
    global vtgate_server
    global vtgate_port

    # Start up a master mysql and vttablet
    logging.debug("Setting up tablets")
    utils.run_vtctl(["CreateKeyspace", KEYSPACE_NAME])
    utils.run_vtctl(["SetKeyspaceShardingInfo", "-force", KEYSPACE_NAME, "keyspace_id", "uint64"])
    shard_0_master.init_tablet("master", keyspace=KEYSPACE_NAME, shard="-80")
    shard_0_replica.init_tablet("replica", keyspace=KEYSPACE_NAME, shard="-80")
    shard_1_master.init_tablet("master", keyspace=KEYSPACE_NAME, shard="80-")
    shard_1_replica.init_tablet("replica", keyspace=KEYSPACE_NAME, shard="80-")

    utils.run_vtctl(["RebuildKeyspaceGraph", KEYSPACE_NAME], auto_log=True)

    for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
        t.create_db("vt_test_keyspace")
        for create_table in create_tables:
            t.mquery(shard_0_master.dbname, create_table)
        t.start_vttablet(wait_for_state=None)

    for t in [shard_0_master, shard_0_replica, shard_1_master, shard_1_replica]:
        t.wait_for_vttablet_state("SERVING")

    utils.run_vtctl(["ReparentShard", "-force", KEYSPACE_NAME + "/-80", shard_0_master.tablet_alias], auto_log=True)
    utils.run_vtctl(["ReparentShard", "-force", KEYSPACE_NAME + "/80-", shard_1_master.tablet_alias], auto_log=True)

    utils.run_vtctl(["RebuildKeyspaceGraph", KEYSPACE_NAME], auto_log=True)

    utils.check_srv_keyspace(
        "test_nj",
        KEYSPACE_NAME,
        "Partitions(master): -80 80-\n" + "Partitions(replica): -80 80-\n" + "TabletTypes: master,replica",
    )

    vtgate_server, vtgate_port = utils.vtgate_start()

#.........这里部分代码省略.........
    shard_2_master.init_tablet('master', 'test_keyspace', '80-c0')
    shard_2_replica1.init_tablet('replica', 'test_keyspace', '80-c0')
    shard_2_replica2.init_tablet('replica', 'test_keyspace', '80-c0')
    shard_2_rdonly1.init_tablet('rdonly', 'test_keyspace', '80-c0')
    shard_3_master.init_tablet('master', 'test_keyspace', 'c0-')
    shard_3_replica.init_tablet('replica', 'test_keyspace', 'c0-')
    shard_3_rdonly1.init_tablet('rdonly', 'test_keyspace', 'c0-')

    # start vttablet on the split shards (no db created,
    # so they're all not serving)
    shard_2_master.start_vttablet(wait_for_state=None)
    shard_3_master.start_vttablet(wait_for_state=None)
    for t in [shard_2_replica1, shard_2_replica2, shard_2_rdonly1,
              shard_3_replica, shard_3_rdonly1]:
      t.start_vttablet(wait_for_state=None)
    for t in [shard_2_master, shard_2_replica1, shard_2_replica2,
              shard_2_rdonly1,
              shard_3_master, shard_3_replica, shard_3_rdonly1]:
      t.wait_for_vttablet_state('NOT_SERVING')

    utils.run_vtctl(['InitShardMaster', 'test_keyspace/80-c0',
                     shard_2_master.tablet_alias], auto_log=True)
    utils.run_vtctl(['InitShardMaster', 'test_keyspace/c0-',
                     shard_3_master.tablet_alias], auto_log=True)

    # check the shards
    shards = utils.run_vtctl_json(['FindAllShardsInKeyspace', 'test_keyspace'])
    for s in ['-80', '80-', '80-c0', 'c0-']:
      self.assertIn(s, shards, 'unexpected shards: %s' % str(shards))
    self.assertEqual(len(shards), 4, 'unexpected shards: %s' % str(shards))

    utils.run_vtctl(['RebuildKeyspaceGraph', 'test_keyspace'],
                    auto_log=True)
    utils.check_srv_keyspace(
        'test_nj', 'test_keyspace',
        'Partitions(master): -80 80-\n'
        'Partitions(rdonly): -80 80-\n'
        'Partitions(replica): -80 80-\n',
        keyspace_id_type=base_sharding.keyspace_id_type,
        sharding_column_name='custom_ksid_col')

    # disable shard_1_slave2, so we're sure filtered replication will go
    # from shard_1_slave1
    utils.run_vtctl(['ChangeSlaveType', shard_1_slave2.tablet_alias, 'spare'])
    shard_1_slave2.wait_for_vttablet_state('NOT_SERVING')

    # we need to create the schema, and the worker will do data copying
    for keyspace_shard in ('test_keyspace/80-c0', 'test_keyspace/c0-'):
      utils.run_vtctl(['CopySchemaShard', '--exclude_tables', 'unrelated',
                       shard_1_rdonly1.tablet_alias, keyspace_shard],
                      auto_log=True)

    # Run vtworker as daemon for the following SplitClone commands.
    worker_proc, worker_port, worker_rpc_port = utils.run_vtworker_bg(
        ['--cell', 'test_nj', '--command_display_interval', '10ms'],
        auto_log=True)

    # Copy the data from the source to the destination shards.
    # min_table_size_for_split is set to 1 as to force a split even on the
    # small table we have.
    # --max_tps is only specified to enable the throttler and ensure that the
    # code is executed. But the intent here is not to throttle the test, hence
    # the rate limit is set very high.
    #
    # Initial clone (online).
    workerclient_proc = utils.run_vtworker_client_bg(

#.........这里部分代码省略.........

    # start vttablet on the split shards (no db created,
    # so they're all not serving)
    sharded_tablets = [shard_0_master, shard_0_replica, shard_0_rdonly1,
                       shard_1_master, shard_1_replica, shard_1_rdonly1]
    for t in sharded_tablets:
      t.wait_for_vttablet_state('NOT_SERVING')

    utils.run_vtctl(['InitShardMaster', '-force', 'test_keyspace/-80',
                     shard_0_master.tablet_alias], auto_log=True)
    utils.run_vtctl(['InitShardMaster', '-force', 'test_keyspace/80-',
                     shard_1_master.tablet_alias], auto_log=True)

    for t in sharded_tablets:
      t.wait_for_vttablet_state('SERVING')

    # must restart vtgate after tablets are up, or else wait until 1min refresh
    # we want cache_ttl at zero so we re-read the topology for every test query.
    utils.vtgate.kill()
    utils.VtGate().start(cache_ttl='0')

    # check the Map Reduce API works correctly, should use ExecuteKeyRanges now,
    # as we are sharded (with just one shard).
    # again, we have 3 values in the database, asking for 4 splits will get us
    # a single query.
    sql = 'select id, msg from resharding1'
    s = utils.vtgate.split_query(sql, 'test_keyspace', 4)
    self.assertEqual(len(s), 1)
    self.assertEqual(s[0]['key_range_part']['keyspace'], 'test_keyspace')
    # There must be one empty KeyRange which represents the full keyspace.
    self.assertEqual(len(s[0]['key_range_part']['key_ranges']), 1)
    self.assertEqual(s[0]['key_range_part']['key_ranges'][0], {})

    utils.check_srv_keyspace('test_nj', 'test_keyspace',
                             'Partitions(master): -\n'
                             'Partitions(rdonly): -\n'
                             'Partitions(replica): -\n',
                             keyspace_id_type=keyspace_id_type)

    # we need to create the schema, and the worker will do data copying
    for keyspace_shard in ('test_keyspace/-80', 'test_keyspace/80-'):
      utils.run_vtctl(['CopySchemaShard',
                       '--exclude_tables', 'unrelated',
                       shard_rdonly1.tablet_alias,
                       keyspace_shard],
                      auto_log=True)
    utils.run_vtctl(['RunHealthCheck', shard_rdonly1.tablet_alias, 'rdonly'])

    utils.run_vtworker(['--cell', 'test_nj',
                        '--command_display_interval', '10ms',
                        'SplitClone',
                        '--exclude_tables', 'unrelated',
                        '--source_reader_count', '10',
                        '--min_table_size_for_split', '1',
                        'test_keyspace/0'],
                       auto_log=True)

    # check the startup values are in the right place
    self._check_startup_values()

    # check the schema too
    utils.run_vtctl(['ValidateSchemaKeyspace', 'test_keyspace'], auto_log=True)

    # check the binlog players are running
    logging.debug('Waiting for binlog players to start on new masters...')
    shard_0_master.wait_for_binlog_player_count(1)

  def test_merge_sharding(self):
    utils.run_vtctl(['CreateKeyspace',
                     '--sharding_column_name', 'custom_sharding_key',
                     '--sharding_column_type', keyspace_id_type,
                     '--split_shard_count', '4',
                     'test_keyspace'])

    shard_0_master.init_tablet('master', 'test_keyspace', '-40')
    shard_0_replica.init_tablet('replica', 'test_keyspace', '-40')
    shard_0_rdonly.init_tablet('rdonly', 'test_keyspace', '-40')
    shard_1_master.init_tablet('master', 'test_keyspace', '40-80')
    shard_1_replica.init_tablet('replica', 'test_keyspace', '40-80')
    shard_1_rdonly.init_tablet('rdonly', 'test_keyspace', '40-80')
    shard_2_master.init_tablet('master', 'test_keyspace', '80-')
    shard_2_replica.init_tablet('replica', 'test_keyspace', '80-')
    shard_2_rdonly.init_tablet('rdonly', 'test_keyspace', '80-')

    utils.run_vtctl(['RebuildKeyspaceGraph', 'test_keyspace'], auto_log=True)

    ks = utils.run_vtctl_json(['GetSrvKeyspace', 'test_nj', 'test_keyspace'])
    self.assertEqual(ks['split_shard_count'], 4)

    # create databases so vttablet can start behaving normally
    for t in [shard_0_master, shard_0_replica, shard_0_rdonly,
              shard_1_master, shard_1_replica, shard_1_rdonly,
              shard_2_master, shard_2_replica, shard_2_rdonly]:
      t.create_db('vt_test_keyspace')
      t.start_vttablet(wait_for_state=None)

    for t in [shard_0_master, shard_0_replica, shard_0_rdonly,
              shard_1_master, shard_1_replica, shard_1_rdonly,
              shard_2_master, shard_2_replica, shard_2_rdonly]:
      t.wait_for_vttablet_state('SERVING')

    # reparent to make the tablets work
    utils.run_vtctl(['InitShardMaster', 'test_keyspace/-40',
                     shard_0_master.tablet_alias], auto_log=True)
    utils.run_vtctl(['InitShardMaster', 'test_keyspace/40-80',
                     shard_1_master.tablet_alias], auto_log=True)
    utils.run_vtctl(['InitShardMaster', 'test_keyspace/80-',
                     shard_2_master.tablet_alias], auto_log=True)

    # create the tables
    self._create_schema()
    self._insert_startup_values()

    # run a health check on source replicas so they respond to discovery
    # (for binlog players) and on the source rdonlys (for workers)
    for t in [shard_0_replica, shard_1_replica]:
      utils.run_vtctl(['RunHealthCheck', t.tablet_alias, 'replica'])
    for t in [shard_0_rdonly, shard_1_rdonly]:
      utils.run_vtctl(['RunHealthCheck', t.tablet_alias, 'rdonly'])

    # create the merge shards
    shard_dest_master.init_tablet('master', 'test_keyspace', '-80')
    shard_dest_replica.init_tablet('replica', 'test_keyspace', '-80')
    shard_dest_rdonly.init_tablet('rdonly', 'test_keyspace', '-80')

    # start vttablet on the split shards (no db created,
    # so they're all not serving)
    for t in [shard_dest_replica, shard_dest_rdonly]:
      t.start_vttablet(wait_for_state=None)
    # Start masters with enabled healthcheck (necessary for resolving the
    # destination master).
    shard_dest_master.start_vttablet(wait_for_state=None,
                                     target_tablet_type='replica')
    for t in [shard_dest_master, shard_dest_replica, shard_dest_rdonly]:
      t.wait_for_vttablet_state('NOT_SERVING')

    utils.run_vtctl(['InitShardMaster', 'test_keyspace/-80',
                     shard_dest_master.tablet_alias], auto_log=True)

    utils.run_vtctl(['RebuildKeyspaceGraph', 'test_keyspace'],
                    auto_log=True)
    utils.check_srv_keyspace(
        'test_nj', 'test_keyspace',
        'Partitions(master): -40 40-80 80-\n'
        'Partitions(rdonly): -40 40-80 80-\n'
        'Partitions(replica): -40 40-80 80-\n',
        keyspace_id_type=keyspace_id_type,
        sharding_column_name='custom_sharding_key')

    # copy the schema
    utils.run_vtctl(['CopySchemaShard', shard_0_rdonly.tablet_alias,
                     'test_keyspace/-80'], auto_log=True)

    # copy the data (will also start filtered replication), reset source
    utils.run_vtworker(['--cell', 'test_nj',
                        '--command_display_interval', '10ms',
                        'SplitClone',
                        '--source_reader_count', '10',
                        '--min_table_size_for_split', '1',
                        '--min_healthy_rdonly_endpoints', '1',
                        'test_keyspace/-80'],
                       auto_log=True)
    utils.run_vtctl(['ChangeSlaveType', shard_0_rdonly.tablet_alias,
                     'rdonly'], auto_log=True)
    utils.run_vtctl(['ChangeSlaveType', shard_1_rdonly.tablet_alias,
                     'rdonly'], auto_log=True)

#.........这里部分代码省略.........

    def test_resharding(self):
        utils.run_vtctl(
            [
                "CreateKeyspace",
                "--sharding_column_name",
                "bad_column",
                "--sharding_column_type",
                "bytes",
                "--split_shard_count",
                "2",
                "test_keyspace",
            ]
        )
        utils.run_vtctl(["SetKeyspaceShardingInfo", "test_keyspace", "keyspace_id", "uint64"], expect_fail=True)
        utils.run_vtctl(
            [
                "SetKeyspaceSharding