def _execute_on_shard(self, query, bind_vars, shard_idx):
    query, bind_vars = dbapi.prepare_query_bind_vars(query, bind_vars)
    for x in xrange(self.max_attempts):
      try:
        conn = self.conns[shard_idx]
        if conn is None:
          conn = self._dial_shard(shard_idx)

        if self.txn:
          self.txn.stmts.append(query)
          if conn not in self.txt.conns:
            # Defer the begin until we actually issue a statement.
            conn.begin()
            self.txt.conns.append(conn)

        return conn._execute(query, bind_vars)
      except dbexceptions.OperationalError as e:
        # Tear down regardless of the precise failure.
        self.conns[shard_idx] = None
        if isinstance(e, tablet3.TimeoutError):
          # On any timeout let the error bubble up and just redial next time.
          raise e

        if isinstance(e, tablet3.RetryError):
          # Give the tablet a moment to restart itself. This isn't
          # strictly necessary since there is a significant chance you
          # will end up talking to another host.
          time.sleep(self.reconnect_delay)
    raise dbexceptions.OperationalError('tablets unreachable', self.keyspace_name, shard_idx, self.db_type)

  def _execute_batch(self, query_list, bind_vars_list, shard_idx):
    new_query_list = []
    new_bind_vars_list = []
    for query, bind_vars in zip(query_list, bind_vars_list):
      query, bind_vars = dbapi.prepare_query_bind_vars(query, bind_vars)
      new_query_list.append(query)
      new_bind_vars_list.append(bind_vars)
    query_list = new_query_list
    bind_vars_list = new_bind_vars_list

    for x in xrange(self.max_attempts):
      try:
        conn = self.conns[shard_idx]
        if conn is None:
          conn = self._dial_shard(shard_idx)

        return conn._execute_batch(query_list, bind_vars_list)
      except dbexceptions.OperationalError as e:
        # Tear down regardless of the precise failure.
        self.conns[shard_idx] = None
        if isinstance(e, tablet3.TimeoutError):
          # On any timeout let the error bubble up and just redial next time.
          raise e

        if isinstance(e, tablet3.RetryError):
          # Give the tablet a moment to restart itself. This isn't
          # strictly necessary since there is a significant chance you
          # will end up talking to another host.
          time.sleep(self.reconnect_delay)
    raise dbexceptions.OperationalError('tablets unreachable', self.keyspace_name, shard_idx, self.db_type)

def _create_v2_request_with_keyspace_ids(sql, new_binds, keyspace_name, tablet_type, keyspace_ids, not_in_transaction):
    """Make a request dict from arguments.

  Args:
    sql: Str sql with format tokens.
    new_binds: Dict of bind variables.
    keyspace_name: Str keyspace name.
    tablet_type: Str tablet_type.
    keyspace_ids: Bytes list of keyspace IDs.
    not_in_transaction: Bool True if a transaction should not be started
      (generally used when sql is not a write).

  Returns:
    A (str: value) dict.
  """
    # keyspace_ids are Keyspace Ids packed to byte[]
    sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
    new_binds = field_types.convert_bind_vars(new_binds)
    req = {
        "Sql": sql,
        "BindVariables": new_binds,
        "Keyspace": keyspace_name,
        "TabletType": topodata_pb2.TabletType.Value(tablet_type.upper()),
        "KeyspaceIds": keyspace_ids,
        "NotInTransaction": not_in_transaction,
    }
    return req

def _create_v2_request_with_keyranges(
    sql, new_binds, keyspace_name, tablet_type, keyranges, not_in_transaction):
  """Make a request dict from arguments.

  Args:
    sql: Str sql with format tokens.
    new_binds: Dict of bind variables.
    keyspace_name: Str keyspace name.
    tablet_type: Str tablet_type.
    keyranges: A list of keyrange.KeyRange objects.
    not_in_transaction: Bool True if a transaction should not be started
      (generally used when sql is not a write).

  Returns:
    A (str: value) dict.
  """
  sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
  new_binds = field_types.convert_bind_vars(new_binds)
  req = {
      'Sql': sql,
      'BindVariables': new_binds,
      'Keyspace': keyspace_name,
      'TabletType': topodata_pb2.TabletType.Value(tablet_type.upper()),
      'KeyRanges': keyranges,
      'NotInTransaction': not_in_transaction,
  }
  return req

 def _execute(self, sql, bind_variables):
   sql, bind_variables = dbapi.prepare_query_bind_vars(sql, bind_variables)
   try:
     result = self.conn._execute(sql, bind_variables)
   except dbexceptions.IntegrityError as e:
     vtdb_logger.get_logger().integrity_error(e)
     raise
   return result

  def _execute_batch(self, sql_list, bind_variables_list):
    sane_sql_list = []
    sane_bind_vars_list = []
    for sql, bind_variables in zip(sql_list, bind_variables_list):
      sane_sql, sane_bind_vars = dbapi.prepare_query_bind_vars(sql, bind_variables)
      sane_sql_list.append(sane_sql)
      sane_bind_vars_list.append(sane_bind_vars)

    result = self.conn._execute_batch(sane_sql_list, sane_bind_vars_list)
    return result

  def _execute_batch(self, sql_list, bind_variables_list):
    sane_sql_list = []
    sane_bind_vars_list = []
    for sql, bind_variables in zip(sql_list, bind_variables_list):
      sane_sql, sane_bind_vars = dbapi.prepare_query_bind_vars(sql, bind_variables)
      sane_sql_list.append(sane_sql)
      sane_bind_vars_list.append(sane_bind_vars)

    result = tablet2.TabletConnection._execute_batch(self, sane_sql_list, sane_bind_vars_list)
    self._time_failed = 0
    return result

  def _execute_batch(
      self, sql_list, bind_variables_list, keyspace_list, keyspace_ids_list,
      tablet_type, as_transaction, effective_caller_id=None):
    query_list = []
    for sql, bind_vars, keyspace, keyspace_ids in zip(
        sql_list, bind_variables_list, keyspace_list, keyspace_ids_list):
      sql, bind_vars = dbapi.prepare_query_bind_vars(sql, bind_vars)
      query = {}
      query['Sql'] = sql
      query['BindVariables'] = field_types.convert_bind_vars(bind_vars)
      query['Keyspace'] = keyspace
      query['KeyspaceIds'] = keyspace_ids
      query_list.append(query)

    rowsets = []

    try:
      req = {
          'Queries': query_list,
          'TabletType': tablet_type,
          'AsTransaction': as_transaction,
      }
      self._add_caller_id(req, effective_caller_id)
      self._add_session(req)
      response = self._get_client().call('VTGate.ExecuteBatchKeyspaceIds', req)
      self._update_session(response)
      if response.reply.get('Error'):
        raise gorpc.AppError(
            response.reply['Error'], 'VTGate.ExecuteBatchKeyspaceIds')
      for reply in response.reply['List']:
        fields = []
        conversions = []
        results = []
        rowcount = 0

        for field in reply['Fields']:
          fields.append((field['Name'], field['Type']))
          conversions.append(field_types.conversions.get(field['Type']))

        for row in reply['Rows']:
          results.append(tuple(_make_row(row, conversions)))

        rowcount = reply['RowsAffected']
        lastrowid = reply['InsertId']
        rowsets.append((results, rowcount, lastrowid, fields))
    except gorpc.GoRpcError as e:
      self.logger_object.log_private_data(bind_variables_list)
      raise self._convert_exception(
          e, sql_list, keyspace_ids_list,
          keyspace='', tablet_type=tablet_type)
    except Exception:
      logging.exception('gorpc low-level error')
      raise
    return rowsets

def _create_req_with_keyranges(sql, new_binds, keyspace, tablet_type, keyranges):
  sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
  new_binds = field_types.convert_bind_vars(new_binds)
  req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'Keyspace': keyspace,
        'TabletType': tablet_type,
        'KeyRanges': [keyrange.KeyRange(kr) for kr in keyranges],
        }
  return req

def _create_req_with_keyspace_ids(sql, new_binds, keyspace, tablet_type, keyspace_ids):
  sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
  new_binds = field_types.convert_bind_vars(new_binds)
  req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'Keyspace': keyspace,
        'TabletType': tablet_type,
        'KeyspaceIds': [str(kid) for kid in keyspace_ids],
        }
  return req

  def _execute_entity_ids(
      self, sql, bind_variables, keyspace, tablet_type,
      entity_keyspace_id_map, entity_column_name, not_in_transaction=False,
      effective_caller_id=None):
    sql, new_binds = dbapi.prepare_query_bind_vars(sql, bind_variables)
    new_binds = field_types.convert_bind_vars(new_binds)
    req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'Keyspace': keyspace,
        'TabletType': tablet_type,
        'EntityKeyspaceIDs': [
            {'ExternalID': xid, 'KeyspaceID': kid}
            for xid, kid in entity_keyspace_id_map.iteritems()],
        'EntityColumnName': entity_column_name,
        'NotInTransaction': not_in_transaction,
        }

    self._add_caller_id(req, effective_caller_id)
    self._add_session(req)

    fields = []
    conversions = []
    results = []
    rowcount = 0
    lastrowid = 0
    try:
      response = self._get_client().call('VTGate.ExecuteEntityIds', req)
      self._update_session(response)
      reply = response.reply
      if response.reply.get('Error'):
        raise gorpc.AppError(response.reply['Error'], 'VTGate.ExecuteEntityIds')

      if reply.get('Result'):
        res = reply['Result']
        for field in res['Fields']:
          fields.append((field['Name'], field['Type']))
          conversions.append(field_types.conversions.get(field['Type']))

        for row in res['Rows']:
          results.append(tuple(_make_row(row, conversions)))

        rowcount = res['RowsAffected']
        lastrowid = res['InsertId']
    except gorpc.GoRpcError as e:
      self.logger_object.log_private_data(bind_variables)
      raise self._convert_exception(
          e, sql, entity_keyspace_id_map,
          keyspace=keyspace, tablet_type=tablet_type)
    except Exception:
      logging.exception('gorpc low-level error')
      raise
    return results, rowcount, lastrowid, fields

def _create_req_with_keyspace_ids(sql, new_binds, keyspace, tablet_type, keyspace_ids):
  # keyspace_ids are Keyspace Ids packed to byte[]
  sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
  new_binds = field_types.convert_bind_vars(new_binds)
  req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'Keyspace': keyspace,
        'TabletType': tablet_type,
        'KeyspaceIds': keyspace_ids,
        }
  return req

def _create_req_with_keyranges(sql, new_binds, keyspace, tablet_type, keyranges):
  # keyranges are keyspace.KeyRange objects with start/end packed to byte[]
  sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
  new_binds = field_types.convert_bind_vars(new_binds)
  req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'Keyspace': keyspace,
        'TabletType': tablet_type,
        'KeyRanges': keyranges,
        }
  return req

def _create_req_with_keyranges(sql, new_binds, keyspace, tablet_type, keyranges, not_in_transaction):
    # keyranges are keyspace.KeyRange objects with start/end packed to byte[]
    sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
    new_binds = field_types.convert_bind_vars(new_binds)
    req = {
        "Sql": sql,
        "BindVariables": new_binds,
        "Keyspace": keyspace,
        "TabletType": tablet_type,
        "KeyRanges": keyranges,
        "NotInTransaction": not_in_transaction,
    }
    return req

  def _execute_batch(self, sql_list, bind_variables_list):
    sane_sql_list = []
    sane_bind_vars_list = []
    for sql, bind_variables in zip(sql_list, bind_variables_list):
      sane_sql, sane_bind_vars = dbapi.prepare_query_bind_vars(sql, bind_variables)
      sane_sql_list.append(sane_sql)
      sane_bind_vars_list.append(sane_bind_vars)

    try:
      result = self.conn._execute_batch(sane_sql_list, sane_bind_vars_list)
    except dbexceptions.IntegrityError as e:
      vtdb_logger.get_logger().integrity_error(e)
      raise

    return result

  def _execute_batch(self, sql_list, bind_variables_list, keyspace, tablet_type, keyspace_ids):
    query_list = []
    for sql, bind_vars in zip(sql_list, bind_variables_list):
      sql, bind_vars = dbapi.prepare_query_bind_vars(sql, bind_vars)
      query = {}
      query['Sql'] = sql
      query['BindVariables'] = field_types.convert_bind_vars(bind_vars)
      query_list.append(query)

    rowsets = []

    try:
      req = {
          'Queries': query_list,
          'Keyspace': keyspace,
          'TabletType': tablet_type,
          'KeyspaceIds': keyspace_ids,
      }
      self._add_session(req)
      response = self.client.call('VTGate.ExecuteBatchKeyspaceIds', req)
      self._update_session(response)
      if 'Error' in response.reply and response.reply['Error']:
        raise gorpc.AppError(response.reply['Error'], 'VTGate.ExecuteBatchKeyspaceIds')
      for reply in response.reply['List']:
        fields = []
        conversions = []
        results = []
        rowcount = 0

        for field in reply['Fields']:
          fields.append((field['Name'], field['Type']))
          conversions.append(field_types.conversions.get(field['Type']))

        for row in reply['Rows']:
          results.append(tuple(_make_row(row, conversions)))

        rowcount = reply['RowsAffected']
        lastrowid = reply['InsertId']
        rowsets.append((results, rowcount, lastrowid, fields))
    except gorpc.GoRpcError as e:
      raise convert_exception(e, str(self), sql_list, bind_variables_list)
    except:
      logging.exception('gorpc low-level error')
      raise
    return rowsets

  def _execute_entity_ids(self, sql, bind_variables, keyspace, tablet_type, entity_keyspace_id_map, entity_column_name):
    sql, new_binds = dbapi.prepare_query_bind_vars(sql, bind_variables)
    new_binds = field_types.convert_bind_vars(new_binds)
    req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'Keyspace': keyspace,
        'TabletType': tablet_type,
        'EntityKeyspaceIDs': [
            {'ExternalID': xid, 'KeyspaceID': kid}
            for xid, kid in entity_keyspace_id_map.iteritems()],
        'EntityColumnName': entity_column_name,
        }

    self._add_session(req)

    fields = []
    conversions = []
    results = []
    try:
      response = self.client.call('VTGate.ExecuteEntityIds', req)
      self._update_session(response)
      reply = response.reply
      if 'Error' in response.reply and response.reply['Error']:
        raise gorpc.AppError(response.reply['Error'], 'VTGate.ExecuteEntityIds')

      if 'Result' in reply:
        res = reply['Result']
        for field in res['Fields']:
          fields.append((field['Name'], field['Type']))
          conversions.append(field_types.conversions.get(field['Type']))

        for row in res['Rows']:
          results.append(tuple(_make_row(row, conversions)))

        rowcount = res['RowsAffected']
        lastrowid = res['InsertId']
    except gorpc.GoRpcError as e:
      raise convert_exception(e, str(self), sql, bind_variables)
    except:
      logging.exception('gorpc low-level error')
      raise
    return results, rowcount, lastrowid, fields

 def build_query_list():
     query_list = []
     for sql, bind_vars, keyspace, keyspace_ids, shards in zip(
         sql_list, bind_variables_list, keyspace_list, keyspace_ids_list, shards_list
     ):
         sql, bind_vars = dbapi.prepare_query_bind_vars(sql, bind_vars)
         query = {}
         query["Sql"] = sql
         query["BindVariables"] = field_types.convert_bind_vars(bind_vars)
         query["Keyspace"] = keyspace
         if keyspace_ids:
             if shards:
                 raise dbexceptions.ProgrammingError(
                     "Keyspace_ids and shards cannot both be defined " "for the same executemany query."
                 )
             query["KeyspaceIds"] = keyspace_ids
         else:
             query["Shards"] = shards
         query_list.append(query)
     return query_list

    def _execute_batch(self, sql_list, bind_variables_list, tablet_type):
        query_list = []
        for sql, bind_vars in zip(sql_list, bind_variables_list):
            sql, bind_vars = dbapi.prepare_query_bind_vars(sql, bind_vars)
            query = {}
            query["Sql"] = sql
            query["BindVariables"] = field_types.convert_bind_vars(bind_vars)
            query_list.append(query)

        rowsets = []

        try:
            req = {"Queries": query_list, "TabletType": tablet_type}
            self._add_session(req)
            response = self.client.call("VTGate.ExecuteBatch", req)
            self._update_session(response)
            if "Error" in response.reply and response.reply["Error"]:
                raise gorpc.AppError(response.reply["Error"], "VTGate.ExecuteBatch")
            for reply in response.reply["List"]:
                fields = []
                conversions = []
                results = []
                rowcount = 0

                for field in reply["Fields"]:
                    fields.append((field["Name"], field["Type"]))
                    conversions.append(field_types.conversions.get(field["Type"]))

                for row in reply["Rows"]:
                    results.append(tuple(_make_row(row, conversions)))

                rowcount = reply["RowsAffected"]
                lastrowid = reply["InsertId"]
                rowsets.append((results, rowcount, lastrowid, fields))
        except gorpc.GoRpcError as e:
            self.logger_object.log_private_data(bind_variables_list)
            raise convert_exception(e, str(self), sql_list)
        except:
            logging.exception("gorpc low-level error")
            raise
        return rowsets

 def build_query_list():
   """Create a query dict list from parameters."""
   query_list = []
   for sql, bind_vars, keyspace_name, keyspace_ids, shards in zip(
       sql_list, bind_variables_list, keyspace_list, keyspace_ids_list,
       shards_list):
     sql, bind_vars = dbapi.prepare_query_bind_vars(sql, bind_vars)
     query = {}
     query['Sql'] = sql
     query['BindVariables'] = field_types.convert_bind_vars(bind_vars)
     query['Keyspace'] = keyspace_name
     if keyspace_ids:
       if shards:
         raise dbexceptions.ProgrammingError(
             'Keyspace_ids and shards cannot both be defined '
             'for the same executemany query.')
       query['KeyspaceIds'] = keyspace_ids
     else:
       query['Shards'] = shards
     query_list.append(query)
   return query_list