// VStreamRows is part of the QueryService interface.
769889
func (itc *internalTabletConn) VStreamRows(ctx context.Context, target *querypb.Target, query string, lastpk *querypb.QueryResult, send func(*binlogdatapb.VStreamRowsResponse) error) error {
	err := itc.tablet.qsc.QueryService().VStreamRows(ctx, target, query, lastpk, send)
	return tabletconn.ErrorFromGRPC(vterrors.ToGRPC(err))
}
384344
func (server *UpdateStream) StreamTables(req *binlogdatapb.StreamTablesRequest, stream binlogservicepb.UpdateStream_StreamTablesServer) (err error) {
	defer server.updateStream.HandlePanic(&err)
	return server.updateStream.StreamTables(stream.Context(), req.Position, req.Tables, req.Charset, func(reply *binlogdatapb.BinlogTransaction) error {
		return stream.Send(&binlogdatapb.StreamTablesResponse{
			BinlogTransaction: reply,
		})
	})
}
1254825
func Rows(v interface{}, r RowsScanner) (outerr error) {
	return rows(v, r, false)
}
848638
func (nuageTable *NuageTable) ReadRows(ovs *libovsdb.OvsdbClient, readRowArgs ReadRowArgs) ([]map[string]interface{}, error) {

	condition := readRowArgs.Condition
	columns := readRowArgs.Columns

	glog.V(2).Infof("Reading rows from table %s with condition (%+v)", nuageTable.TableName, condition)

	var selectOp libovsdb.Operation

	if len(condition) == 3 {

		ovsdbCondition := libovsdb.NewCondition(condition[0], condition[1], condition[2])

		if columns == nil {
			selectOp = libovsdb.Operation{
				Op:    "select",
				Table: nuageTable.TableName,
				Where: []interface{}{ovsdbCondition},
			}
		} else {
			selectOp = libovsdb.Operation{
				Op:      "select",
				Table:   nuageTable.TableName,
				Where:   []interface{}{ovsdbCondition},
				Columns: columns,
			}
		}
	} else {
		if columns == nil {
			selectOp = libovsdb.Operation{
				Op:    "select",
				Table: nuageTable.TableName,
				Where: []interface{}{},
			}
		} else {
			selectOp = libovsdb.Operation{
				Op:      "select",
				Table:   nuageTable.TableName,
				Columns: columns,
				Where:   []interface{}{},
			}
		}
	}

	operations := []libovsdb.Operation{selectOp}
	reply, err := ovs.Transact(OvsDBName, operations...)

	glog.V(2).Infof("reply : (%+v) err : (%+v)", reply, err)

	if err != nil || len(reply) != 1 || reply[0].Error != "" {
		glog.Errorf("Problem reading row from the Nuage table %s %v %+v", nuageTable.TableName, err, reply)
		return nil, fmt.Errorf("Problem reading row from the Nuage table %s %v",
			nuageTable.TableName, err)
	}

	return reply[0].Rows, nil
}
338071
func (sbc *SandboxConn) VStreamRows(ctx context.Context, target *querypb.Target, query string, lastpk *querypb.QueryResult, send func(*binlogdatapb.VStreamRowsResponse) error) error {
	return fmt.Errorf("not implemented in test")
}
667651
func (vs *vstreamer) Stream() error {
	defer vs.cancel()

	pos, err := mysql.DecodePosition(vs.startPos)
	if err != nil {
		return err
	}
	vs.pos = pos

	// Ensure se is Open. If vttablet came up in a non_serving role,
	// the schema engine may not have been initialized.
	if err := vs.se.Open(); err != nil {
		return wrapError(err, vs.pos)
	}

	conn, err := binlog.NewSlaveConnection(vs.cp)
	if err != nil {
		return wrapError(err, vs.pos)
	}
	defer conn.Close()

	events, err := conn.StartBinlogDumpFromPosition(vs.ctx, vs.pos)
	if err != nil {
		return wrapError(err, vs.pos)
	}
	err = vs.parseEvents(vs.ctx, events)
	return wrapError(err, vs.pos)
}
1278238
func NewResultStream(streams ...ResultStream) ResultStream {
	switch len(streams) {
	case 0:
		return nil
	case 1:
		return streams[0]
	default:
	}

	linkResult(streams[0], streams[1])

	return struct {
		ResultConsumer
		ResultProducer
	}{
		streams[0],
		NewResultStream(streams[1:]...),
	}
}
298641
func (client *QueryClient) StreamExecuteWithOptions(query string, bindvars map[string]*querypb.BindVariable, options *querypb.ExecuteOptions) (*sqltypes.Result, error) {
	result := &sqltypes.Result{}
	err := client.server.StreamExecute(
		client.ctx,
		&client.target,
		query,
		bindvars,
		0,
		options,
		func(res *sqltypes.Result) error {
			if result.Fields == nil {
				result.Fields = res.Fields
			}
			result.Rows = append(result.Rows, res.Rows...)
			result.RowsAffected += uint64(len(res.Rows))
			return nil
		},
	)
	if err != nil {
		return nil, err
	}
	return result, nil
}
75594
func (conn *FakeVTGateConn) StreamExecute(ctx context.Context, session *vtgatepb.Session, sql string, bindVars map[string]*querypb.BindVariable) (sqltypes.ResultStream, error) {
	response, ok := conn.execMap[sql]
	if !ok {
		return nil, fmt.Errorf("no match for: %s", sql)
	}
	query := &queryExecute{
		SQL:           sql,
		BindVariables: bindVars,
		Session:       session,
	}
	if !reflect.DeepEqual(query, response.execQuery) {
		return nil, fmt.Errorf("StreamExecute: %+v, want %+v", sql, response.execQuery)
	}
	if response.err != nil {
		return nil, response.err
	}
	var resultChan chan *sqltypes.Result
	defer close(resultChan)
	if response.reply != nil {
		// create a result channel big enough to buffer all of
		// the responses so we don't need to fork a go routine.
		resultChan = make(chan *sqltypes.Result, len(response.reply.Rows)+1)
		result := &sqltypes.Result{}
		result.Fields = response.reply.Fields
		resultChan <- result
		for _, row := range response.reply.Rows {
			result := &sqltypes.Result{}
			result.Rows = [][]sqltypes.Value{row}
			resultChan <- result
		}
	} else {
		resultChan = make(chan *sqltypes.Result)
	}
	return &streamExecuteAdapter{resultChan}, nil
}
459454
func (c *Chunk) NumRows() int {
	if c.NumCols() == 0 {
		return c.numVirtualRows
	}
	return c.columns[0].length
}
828703
func (conn *FakeVTGateConn) StreamExecuteShards(ctx context.Context, query string, keyspace string, shards []string, bindVars map[string]*querypb.BindVariable, tabletType topodatapb.TabletType, options *querypb.ExecuteOptions) (sqltypes.ResultStream, error) {
	panic("not implemented")
}
23430
func (c *ExtensionManagerClient) QueryRows(sql string) ([]map[string]string, error) {
	res, err := c.Query(sql)
	if err != nil {
		return nil, errors.Wrap(err, "transport error in query")
	}
	if res.Status == nil {
		return nil, errors.New("query returned nil status")
	}
	if res.Status.Code != 0 {
		return nil, errors.Errorf("query returned error: %s", res.Status.Message)
	}
	return res.Response, nil

}
801655
func (me *TxDatabase) QueryRow(query string, args ...interface{}) *sql.Row {
	me.Trace("QUERY ROW", query, args...)
	return me.Tx.QueryRow(query, args...)
}
880197
func (c *Conn) ExecuteStreamFetch(query string) (err error) {
	defer func() {
		if err != nil {
			if sqlerr, ok := err.(*SQLError); ok {
				sqlerr.Query = query
			}
		}
	}()

	// Sanity check.
	if c.fields != nil {
		return NewSQLError(CRCommandsOutOfSync, SSUnknownSQLState, "streaming query already in progress")
	}

	// Send the query as a COM_QUERY packet.
	if err := c.WriteComQuery(query); err != nil {
		return err
	}

	// Get the result.
	_, _, colNumber, _, _, err := c.readComQueryResponse()
	if err != nil {
		return err
	}
	if colNumber == 0 {
		// OK packet, means no results. Save an empty Fields array.
		c.fields = make([]*querypb.Field, 0)
		return nil
	}

	// Read the fields, save them.
	fields := make([]querypb.Field, colNumber)
	fieldsPointers := make([]*querypb.Field, colNumber)

	// Read column headers. One packet per column.
	// Build the fields.
	for i := 0; i < colNumber; i++ {
		fieldsPointers[i] = &fields[i]
		if err := c.readColumnDefinition(fieldsPointers[i], i); err != nil {
			return err
		}
	}

	// Read the EOF after the fields if necessary.
	if c.Capabilities&CapabilityClientDeprecateEOF == 0 {
		// EOF is only present here if it's not deprecated.
		data, err := c.readEphemeralPacket()
		if err != nil {
			return NewSQLError(CRServerLost, SSUnknownSQLState, "%v", err)
		}
		defer c.recycleReadPacket()
		if isEOFPacket(data) {
			// This is what we expect.
			// Warnings and status flags are ignored.
			// goto: end
		} else if isErrorPacket(data) {
			return ParseErrorPacket(data)
		} else {
			return NewSQLError(CRCommandsOutOfSync, SSUnknownSQLState, "unexpected packet after fields: %v", data)
		}
	}

	c.fields = fieldsPointers
	return nil
}
679967
func (tsv *TabletServer) VStreamRows(ctx context.Context, target *querypb.Target, query string, lastpk *querypb.QueryResult, send func(*binlogdatapb.VStreamRowsResponse) error) error {
	if err := tsv.verifyTarget(ctx, target); err != nil {
		return err
	}
	var row []sqltypes.Value
	if lastpk != nil {
		r := sqltypes.Proto3ToResult(lastpk)
		if len(r.Rows) != 1 {
			return vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "unexpected lastpk input: %v", lastpk)
		}
		row = r.Rows[0]
	}
	return tsv.vstreamer.StreamRows(ctx, query, row, send)
}
1064843
func (route *Route) StreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
	var rss []*srvtopo.ResolvedShard
	var bvs []map[string]*querypb.BindVariable
	var err error
	if route.QueryTimeout != 0 {
		cancel := vcursor.SetContextTimeout(time.Duration(route.QueryTimeout) * time.Millisecond)
		defer cancel()
	}
	switch route.Opcode {
	case SelectUnsharded, SelectNext, SelectDBA, SelectReference:
		rss, bvs, err = route.paramsAnyShard(vcursor, bindVars)
	case SelectScatter:
		rss, bvs, err = route.paramsAllShards(vcursor, bindVars)
	case SelectEqual, SelectEqualUnique:
		rss, bvs, err = route.paramsSelectEqual(vcursor, bindVars)
	case SelectIN:
		rss, bvs, err = route.paramsSelectIn(vcursor, bindVars)
	default:
		return fmt.Errorf("query %q cannot be used for streaming", route.Query)
	}
	if err != nil {
		return err
	}

	// No route.
	if len(rss) == 0 {
		if wantfields {
			r, err := route.GetFields(vcursor, bindVars)
			if err != nil {
				return err
			}
			return callback(r)
		}
		return nil
	}

	if len(route.OrderBy) == 0 {
		return vcursor.StreamExecuteMulti(route.Query, rss, bvs, func(qr *sqltypes.Result) error {
			return callback(qr.Truncate(route.TruncateColumnCount))
		})
	}

	return mergeSort(vcursor, route.Query, route.OrderBy, rss, bvs, func(qr *sqltypes.Result) error {
		return callback(qr.Truncate(route.TruncateColumnCount))
	})
}
7500
func (tx *Tx) QueryRow(query string, args ...interface{}) *Row {
	rows, err := tx.Query(query, args...)
	return &Row{rows: rows, err: err}
}