// BeginExecute starts a transaction and runs an Execute.
func (conn *gRPCQueryClient) BeginExecute(ctx context.Context, target *querypb.Target, query string, bindVars map[string]interface{}, options *querypb.ExecuteOptions) (result *sqltypes.Result, transactionID int64, err error) {
	conn.mu.RLock()
	defer conn.mu.RUnlock()
	if conn.cc == nil {
		return nil, 0, tabletconn.ConnClosed
	}

	q, err := querytypes.BoundQueryToProto3(query, bindVars)
	if err != nil {
		return nil, 0, err
	}

	if *combo {
		// If combo is enabled, we combine both calls
		req := &querypb.BeginExecuteRequest{
			Target:            target,
			EffectiveCallerId: callerid.EffectiveCallerIDFromContext(ctx),
			ImmediateCallerId: callerid.ImmediateCallerIDFromContext(ctx),
			Query:             q,
			Options:           options,
		}
		reply, err := conn.c.BeginExecute(ctx, req)
		if err != nil {
			return nil, 0, tabletconn.TabletErrorFromGRPC(err)
		}
		if reply.Error != nil {
			return nil, reply.TransactionId, tabletconn.TabletErrorFromRPCError(reply.Error)
		}
		return sqltypes.Proto3ToResult(reply.Result), reply.TransactionId, nil
	}

	// Begin part.
	breq := &querypb.BeginRequest{
		Target:            target,
		EffectiveCallerId: callerid.EffectiveCallerIDFromContext(ctx),
		ImmediateCallerId: callerid.ImmediateCallerIDFromContext(ctx),
	}
	br, err := conn.c.Begin(ctx, breq)
	if err != nil {
		return nil, 0, tabletconn.TabletErrorFromGRPC(err)
	}
	transactionID = br.TransactionId

	// Execute part.
	ereq := &querypb.ExecuteRequest{
		Target:            target,
		EffectiveCallerId: breq.EffectiveCallerId,
		ImmediateCallerId: breq.ImmediateCallerId,
		Query:             q,
		TransactionId:     transactionID,
		Options:           options,
	}
	er, err := conn.c.Execute(ctx, ereq)
	if err != nil {
		return nil, transactionID, tabletconn.TabletErrorFromGRPC(err)
	}
	return sqltypes.Proto3ToResult(er.Result), transactionID, nil

}

// copyShardMetadata copies contents of _vt.shard_metadata table from the source
// tablet to the destination tablet. It's assumed that destination tablet is a
// master and binlogging is not turned off when INSERT statements are executed.
func (wr *Wrangler) copyShardMetadata(ctx context.Context, srcTabletAlias *topodatapb.TabletAlias, destTabletAlias *topodatapb.TabletAlias) error {
	presenceResult, err := wr.ExecuteFetchAsDba(ctx, srcTabletAlias, "SELECT 1 FROM information_schema.tables WHERE table_schema = '_vt' AND table_name = 'shard_metadata'", 1, false, false)
	if err != nil {
		return err
	}
	if len(presenceResult.Rows) == 0 {
		log.Infof("_vt.shard_metadata doesn't exist on the source tablet %v, skipping its copy.", topoproto.TabletAliasString(srcTabletAlias))
		return nil
	}

	dataProto, err := wr.ExecuteFetchAsDba(ctx, srcTabletAlias, "SELECT name, value FROM _vt.shard_metadata", 100, false, false)
	if err != nil {
		return err
	}
	data := sqltypes.Proto3ToResult(dataProto)
	for _, row := range data.Rows {
		name := row[0]
		value := row[1]
		queryBuf := bytes.Buffer{}
		queryBuf.WriteString("INSERT INTO _vt.shard_metadata (name, value) VALUES (")
		name.EncodeSQL(&queryBuf)
		queryBuf.WriteByte(',')
		value.EncodeSQL(&queryBuf)
		queryBuf.WriteString(") ON DUPLICATE KEY UPDATE value = ")
		value.EncodeSQL(&queryBuf)

		_, err := wr.ExecuteFetchAsDba(ctx, destTabletAlias, queryBuf.String(), 0, false, false)
		if err != nil {
			return err
		}
	}
	return nil
}

func (conn *vtgateConn) ExecuteShards(ctx context.Context, query string, keyspace string, shards []string, bindVars map[string]interface{}, tabletType topodatapb.TabletType, session interface{}) (*sqltypes.Result, interface{}, error) {
	var s *vtgatepb.Session
	if session != nil {
		s = session.(*vtgatepb.Session)
	}
	q, err := querytypes.BoundQueryToProto3(query, bindVars)
	if err != nil {
		return nil, session, err
	}
	request := &vtgatepb.ExecuteShardsRequest{
		CallerId:   callerid.EffectiveCallerIDFromContext(ctx),
		Session:    s,
		Query:      q,
		Keyspace:   keyspace,
		Shards:     shards,
		TabletType: tabletType,
	}
	response, err := conn.c.ExecuteShards(ctx, request)
	if err != nil {
		return nil, session, vterrors.FromGRPCError(err)
	}
	if response.Error != nil {
		return nil, response.Session, vterrors.FromVtRPCError(response.Error)
	}
	return sqltypes.Proto3ToResult(response.Result), response.Session, nil
}

func (conn *vtgateConn) Execute(ctx context.Context, query string, bindVars map[string]interface{}, tabletType topodatapb.TabletType, notInTransaction bool, session interface{}) (*sqltypes.Result, interface{}, error) {
	var s *pb.Session
	if session != nil {
		s = session.(*pb.Session)
	}
	q, err := tproto.BoundQueryToProto3(query, bindVars)
	if err != nil {
		return nil, session, err
	}
	request := &pb.ExecuteRequest{
		CallerId:         callerid.EffectiveCallerIDFromContext(ctx),
		Session:          s,
		Query:            q,
		TabletType:       tabletType,
		NotInTransaction: notInTransaction,
	}
	response, err := conn.c.Execute(ctx, request)
	if err != nil {
		return nil, session, vterrors.FromGRPCError(err)
	}
	if response.Error != nil {
		return nil, response.Session, vterrors.FromVtRPCError(response.Error)
	}
	return sqltypes.Proto3ToResult(response.Result), response.Session, nil
}

func (conn *vtgateConn) StreamExecuteKeyspaceIds(ctx context.Context, query string, keyspace string, keyspaceIds [][]byte, bindVars map[string]interface{}, tabletType topodatapb.TabletType) (<-chan *sqltypes.Result, vtgateconn.ErrFunc, error) {
	q, err := tproto.BoundQueryToProto3(query, bindVars)
	if err != nil {
		return nil, nil, err
	}
	req := &pb.StreamExecuteKeyspaceIdsRequest{
		CallerId:    callerid.EffectiveCallerIDFromContext(ctx),
		Query:       q,
		Keyspace:    keyspace,
		KeyspaceIds: keyspaceIds,
		TabletType:  tabletType,
	}
	stream, err := conn.c.StreamExecuteKeyspaceIds(ctx, req)
	if err != nil {
		return nil, nil, vterrors.FromGRPCError(err)
	}
	sr := make(chan *sqltypes.Result, 10)
	var finalError error
	go func() {
		for {
			ser, err := stream.Recv()
			if err != nil {
				if err != io.EOF {
					finalError = vterrors.FromGRPCError(err)
				}
				close(sr)
				return
			}
			sr <- sqltypes.Proto3ToResult(ser.Result)
		}
	}()
	return sr, func() error {
		return finalError
	}, nil
}

// Execute sends the query to VTTablet.
func (conn *gRPCQueryClient) Execute(ctx context.Context, query string, bindVars map[string]interface{}, transactionID int64) (*sqltypes.Result, error) {
	conn.mu.RLock()
	defer conn.mu.RUnlock()
	if conn.cc == nil {
		return nil, tabletconn.ConnClosed
	}

	q, err := tproto.BoundQueryToProto3(query, bindVars)
	if err != nil {
		return nil, err
	}

	req := &pb.ExecuteRequest{
		Target:            conn.target,
		EffectiveCallerId: callerid.EffectiveCallerIDFromContext(ctx),
		ImmediateCallerId: callerid.ImmediateCallerIDFromContext(ctx),
		Query:             q,
		TransactionId:     transactionID,
		SessionId:         conn.sessionID,
	}
	er, err := conn.c.Execute(ctx, req)
	if err != nil {
		return nil, tabletconn.TabletErrorFromGRPC(err)
	}
	return sqltypes.Proto3ToResult(er.Result), nil
}

// executeAdminQuery executes a query on a given tablet as 'allprivs' user. The query is executed
// using timeout value from --schema_swap_admin_query_timeout flag.
func (shardSwap *shardSchemaSwap) executeAdminQuery(tablet *topodatapb.Tablet, query string, maxRows int) (*sqltypes.Result, error) {
	sqlCtx, cancelSQLCtx := context.WithTimeout(shardSwap.parent.ctx, *adminQueryTimeout)
	defer cancelSQLCtx()

	sqlResultProto, err := shardSwap.parent.tabletClient.ExecuteFetchAsAllPrivs(
		sqlCtx,
		tablet,
		[]byte(query),
		maxRows,
		false /* reloadSchema */)
	if err != nil {
		return nil, err
	}
	return sqltypes.Proto3ToResult(sqlResultProto), nil
}

// ExecuteFetchAsApp is part of the tmclient.TabletManagerClient interface
func (client *Client) ExecuteFetchAsApp(ctx context.Context, tablet *topo.TabletInfo, query string, maxRows int, wantFields bool) (*sqltypes.Result, error) {
	cc, c, err := client.dial(ctx, tablet)
	if err != nil {
		return nil, err
	}
	defer cc.Close()
	response, err := c.ExecuteFetchAsApp(ctx, &tabletmanagerdatapb.ExecuteFetchAsAppRequest{
		Query:      query,
		MaxRows:    uint64(maxRows),
		WantFields: wantFields,
	})
	if err != nil {
		return nil, err
	}
	return sqltypes.Proto3ToResult(response.Result), nil
}

// ExecuteFetchAsDba is part of the tmclient.TabletManagerClient interface
func (client *Client) ExecuteFetchAsDba(ctx context.Context, tablet *topo.TabletInfo, query string, maxRows int, wantFields, disableBinlogs, reloadSchema bool) (*sqltypes.Result, error) {
	cc, c, err := client.dial(ctx, tablet)
	if err != nil {
		return nil, err
	}
	defer cc.Close()
	response, err := c.ExecuteFetchAsDba(ctx, &tabletmanagerdatapb.ExecuteFetchAsDbaRequest{
		Query:          query,
		DbName:         tablet.DbName(),
		MaxRows:        uint64(maxRows),
		WantFields:     wantFields,
		DisableBinlogs: disableBinlogs,
		ReloadSchema:   reloadSchema,
	})
	if err != nil {
		return nil, err
	}
	return sqltypes.Proto3ToResult(response.Result), nil
}

// StreamExecute starts a streaming query to VTTablet.
func (conn *gRPCQueryClient) StreamExecute(ctx context.Context, query string, bindVars map[string]interface{}, transactionID int64) (<-chan *sqltypes.Result, tabletconn.ErrFunc, error) {
	conn.mu.RLock()
	defer conn.mu.RUnlock()
	if conn.cc == nil {
		return nil, nil, tabletconn.ConnClosed
	}

	q, err := tproto.BoundQueryToProto3(query, bindVars)
	if err != nil {
		return nil, nil, err
	}
	req := &pb.StreamExecuteRequest{
		Target:            conn.target,
		EffectiveCallerId: callerid.EffectiveCallerIDFromContext(ctx),
		ImmediateCallerId: callerid.ImmediateCallerIDFromContext(ctx),
		Query:             q,
		SessionId:         conn.sessionID,
	}
	stream, err := conn.c.StreamExecute(ctx, req)
	if err != nil {
		return nil, nil, tabletconn.TabletErrorFromGRPC(err)
	}
	sr := make(chan *sqltypes.Result, 10)
	var finalError error
	go func() {
		for {
			ser, err := stream.Recv()
			if err != nil {
				if err != io.EOF {
					finalError = tabletconn.TabletErrorFromGRPC(err)
				}
				close(sr)
				return
			}
			sr <- sqltypes.Proto3ToResult(ser.Result)
		}
	}()
	return sr, func() error {
		return finalError
	}, nil
}

// generateChunks returns an array of chunks to use for splitting up a table
// into multiple data chunks. It only works for tables with a primary key
// whose first column is a numeric type.
func generateChunks(ctx context.Context, wr *wrangler.Wrangler, tablet *topodatapb.Tablet, td *tabletmanagerdatapb.TableDefinition, chunkCount, minRowsPerChunk int) ([]chunk, error) {
	if len(td.PrimaryKeyColumns) == 0 {
		// No explicit primary key. Cannot chunk the rows then.
		wr.Logger().Infof("table=%v: Not splitting the table into multiple chunks because it has no primary key columns. This will reduce the performance of the clone.", td.Name)
		return singleCompleteChunk, nil
	}
	if td.RowCount < 2*uint64(minRowsPerChunk) {
		// The automatic adjustment of "chunkCount" based on "minRowsPerChunk"
		// below would set "chunkCount" to less than 2 i.e. 1 or 0 chunks.
		// In practice in this case there should be exactly one chunk.
		// Return early in this case and notice the user about this.
		wr.Logger().Infof("table=%v: Not splitting the table into multiple chunks because it has only %d rows.", td.Name, td.RowCount)
		return singleCompleteChunk, nil
	}
	if chunkCount == 1 {
		return singleCompleteChunk, nil
	}

	// Get the MIN and MAX of the leading column of the primary key.
	query := fmt.Sprintf("SELECT MIN(%v), MAX(%v) FROM %v.%v", escape(td.PrimaryKeyColumns[0]), escape(td.PrimaryKeyColumns[0]), escape(topoproto.TabletDbName(tablet)), escape(td.Name))
	shortCtx, cancel := context.WithTimeout(ctx, *remoteActionsTimeout)
	qr, err := wr.TabletManagerClient().ExecuteFetchAsApp(shortCtx, tablet, true, []byte(query), 1)
	cancel()
	if err != nil {
		return nil, fmt.Errorf("Cannot determine MIN and MAX of the first primary key column. ExecuteFetchAsApp: %v", err)
	}
	if len(qr.Rows) != 1 {
		return nil, fmt.Errorf("Cannot determine MIN and MAX of the first primary key column. Zero rows were returned for the following query: %v", query)
	}

	result := sqltypes.Proto3ToResult(qr)
	min := result.Rows[0][0].ToNative()
	max := result.Rows[0][1].ToNative()

	if min == nil || max == nil {
		wr.Logger().Infof("table=%v: Not splitting the table into multiple chunks, min or max is NULL: %v", td.Name, qr.Rows[0])
		return singleCompleteChunk, nil
	}

	// Determine the average number of rows per chunk for the given chunkCount.
	avgRowsPerChunk := td.RowCount / uint64(chunkCount)
	if avgRowsPerChunk < uint64(minRowsPerChunk) {
		// Reduce the chunkCount to fulfill minRowsPerChunk.
		newChunkCount := td.RowCount / uint64(minRowsPerChunk)
		wr.Logger().Infof("table=%v: Reducing the number of chunks from the default %d to %d to make sure that each chunk has at least %d rows.", td.Name, chunkCount, newChunkCount, minRowsPerChunk)
		chunkCount = int(newChunkCount)
	}

	// TODO(mberlin): Write a unit test for this part of the function.
	var interval interface{}
	chunks := make([]chunk, chunkCount)
	switch min := min.(type) {
	case int64:
		max := max.(int64)
		interval = (max - min) / int64(chunkCount)
		if interval == 0 {
			wr.Logger().Infof("table=%v: Not splitting the table into multiple chunks, interval=0: %v to %v", td.Name, min, max)
			return singleCompleteChunk, nil
		}
	case uint64:
		max := max.(uint64)
		interval = (max - min) / uint64(chunkCount)
		if interval == 0 {
			wr.Logger().Infof("table=%v: Not splitting the table into multiple chunks, interval=0: %v to %v", td.Name, min, max)
			return singleCompleteChunk, nil
		}
	case float64:
		max := max.(float64)
		interval = (max - min) / float64(chunkCount)
		if interval == 0 {
			wr.Logger().Infof("table=%v: Not splitting the table into multiple chunks, interval=0: %v to %v", td.Name, min, max)
			return singleCompleteChunk, nil
		}
	default:
		wr.Logger().Infof("table=%v: Not splitting the table into multiple chunks, primary key not numeric.", td.Name)
		return singleCompleteChunk, nil
	}

	// Create chunks.
	start := min
	for i := 0; i < chunkCount; i++ {
		end := add(start, interval)
		chunk, err := toChunk(start, end, i+1, chunkCount)
		if err != nil {
			return nil, err
		}
		chunks[i] = chunk
		start = end
	}

	// Clear out the MIN and MAX on the first and last chunk respectively
	// because other shards might have smaller or higher values than the one we
	// looked at.
	chunks[0].start = sqltypes.NULL
	chunks[chunkCount-1].end = sqltypes.NULL
	return chunks, nil
}

// generateChunks returns an array of chunks to use for splitting up a table
// into multiple data chunks. It only works for tables with a primary key
// whose first column is a numeric type.
func generateChunks(ctx context.Context, wr *wrangler.Wrangler, tablet *topodatapb.Tablet, td *tabletmanagerdatapb.TableDefinition, minTableSizeForSplit uint64, chunkCount int) ([]chunk, error) {
	if len(td.PrimaryKeyColumns) == 0 {
		// No explicit primary key. Cannot chunk the rows then.
		return singleCompleteChunk, nil
	}
	if td.DataLength < minTableSizeForSplit {
		// Table is too small to split up.
		return singleCompleteChunk, nil
	}
	if chunkCount == 1 {
		return singleCompleteChunk, nil
	}

	// Get the MIN and MAX of the leading column of the primary key.
	query := fmt.Sprintf("SELECT MIN(%v), MAX(%v) FROM %v.%v", td.PrimaryKeyColumns[0], td.PrimaryKeyColumns[0], topoproto.TabletDbName(tablet), td.Name)
	shortCtx, cancel := context.WithTimeout(ctx, *remoteActionsTimeout)
	qr, err := wr.TabletManagerClient().ExecuteFetchAsApp(shortCtx, tablet, true, []byte(query), 1)
	cancel()
	if err != nil {
		return nil, fmt.Errorf("Cannot determine MIN and MAX of the first primary key column. ExecuteFetchAsApp: %v", err)
	}
	if len(qr.Rows) != 1 {
		return nil, fmt.Errorf("Cannot determine MIN and MAX of the first primary key column. Zero rows were returned for the following query: %v", query)
	}

	result := sqltypes.Proto3ToResult(qr)
	min := result.Rows[0][0].ToNative()
	max := result.Rows[0][1].ToNative()

	if min == nil || max == nil {
		wr.Logger().Infof("Not splitting table %v into multiple chunks, min or max is NULL: %v", td.Name, qr.Rows[0])
		return singleCompleteChunk, nil
	}

	// TODO(mberlin): Write a unit test for this part of the function.
	chunks := make([]chunk, chunkCount)
	switch min := min.(type) {
	case int64:
		max := max.(int64)
		interval := (max - min) / int64(chunkCount)
		if interval == 0 {
			wr.Logger().Infof("Not splitting table %v into multiple chunks, interval=0: %v to %v", td.Name, min, max)
			return singleCompleteChunk, nil
		}

		start := min
		for i := 0; i < chunkCount; i++ {
			end := start + interval
			chunk, err := toChunk(start, end)
			if err != nil {
				return nil, err
			}
			chunks[i] = chunk
			start = end
		}
	case uint64:
		max := max.(uint64)
		interval := (max - min) / uint64(chunkCount)
		if interval == 0 {
			wr.Logger().Infof("Not splitting table %v into multiple chunks, interval=0: %v to %v", td.Name, min, max)
			return singleCompleteChunk, nil
		}

		start := min
		for i := 0; i < chunkCount; i++ {
			end := start + interval
			chunk, err := toChunk(start, end)
			if err != nil {
				return nil, err
			}
			chunks[i] = chunk
			start = end
		}
	case float64:
		max := max.(float64)
		interval := (max - min) / float64(chunkCount)
		if interval == 0 {
			wr.Logger().Infof("Not splitting table %v into multiple chunks, interval=0: %v to %v", td.Name, min, max)
			return singleCompleteChunk, nil
		}

		start := min
		for i := 0; i < chunkCount; i++ {
			end := start + interval
			chunk, err := toChunk(start, end)
			if err != nil {
				return nil, err
			}
			chunks[i] = chunk
			start = end
		}
	default:
		wr.Logger().Infof("Not splitting table %v into multiple chunks, primary key not numeric.", td.Name)
		return singleCompleteChunk, nil
	}

	// Clear out the MIN and MAX on the first and last chunk respectively
//.........这里部分代码省略.........