// VStreamRows is part of the QueryService interface.
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")
}
472014
func (cdc *ChannelDriverConn) ReadRows() (chan []string, error) {
	if cdc.conn == nil {
		return nil, fmt.Errorf("connection closed")
	}

	m := <-cdc.conn
	if m.Type == errMessage {
		return nil, errors.New(m.Value[0])
	}

	if m.Type != rowHeaderMessage {
		return nil, errors.New("not a rows header")
	}

	return UnlimitedRowsChannel(cdc.conn, m), nil
}
1158920
func (l *Client) GetStreams(stream string) ([]*Stream, error) {
	res := ApiResp{}
	data := []*Stream{}

	// make the request
	err := l.Get(parseLyticsURL(routeSchemaStreams, nil), nil, nil, &res, &data)
	if err != nil {
		return data, err
	}

	return data, nil
}
459454
func (c *Chunk) NumRows() int {
	if c.NumCols() == 0 {
		return c.numVirtualRows
	}
	return c.columns[0].length
}
1034468
func (e *InsertValues) insertRows(ctx context.Context, exec func(ctx context.Context, rows [][]types.Datum) error) (err error) {
	// For `insert|replace into set x=y`, process the set list here.
	if err = e.processSetList(); err != nil {
		return err
	}
	sessVars := e.ctx.GetSessionVars()
	batchInsert := sessVars.BatchInsert && !sessVars.InTxn()
	batchSize := sessVars.DMLBatchSize

	rows := make([][]types.Datum, 0, len(e.Lists))
	for i, list := range e.Lists {
		e.rowCount++
		row, err := e.evalRow(list, i)
		if err != nil {
			return err
		}
		rows = append(rows, row)
		if batchInsert && e.rowCount%uint64(batchSize) == 0 {
			if err = exec(ctx, rows); err != nil {
				return err
			}
			rows = rows[:0]
			if err = e.doBatchInsert(ctx); err != nil {
				return err
			}
		}
	}
	return exec(ctx, rows)
}
1208986
func (tsv *TabletServer) VStream(ctx context.Context, target *querypb.Target, startPos string, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) error {
	if err := tsv.verifyTarget(ctx, target); err != nil {
		return err
	}
	return tsv.vstreamer.Stream(ctx, startPos, filter, send)
}
551314
func NewResultRows(ctx *plan.Context, cols []string) *ResultWriter {
	stepper := NewTaskStepper(ctx)
	m := &ResultWriter{
		TaskBase: stepper.TaskBase,
		cols:     cols,
	}
	return m
}
1092302
func (vtg *VTGate) StreamExecuteShards(request *vtgatepb.StreamExecuteShardsRequest, stream vtgateservicepb.Vitess_StreamExecuteShardsServer) (err error) {
	defer vtg.server.HandlePanic(&err)
	ctx := withCallerIDContext(stream.Context(), request.CallerId)
	vtgErr := vtg.server.StreamExecuteShards(ctx,
		request.Query.Sql,
		request.Query.BindVariables,
		request.Keyspace,
		request.Shards,
		request.TabletType,
		request.Options,
		func(value *sqltypes.Result) error {
			// Send is not safe to call concurrently, but vtgate
			// guarantees that it's not.
			return stream.Send(&vtgatepb.StreamExecuteShardsResponse{
				Result: sqltypes.ResultToProto3(value),
			})
		})
	return vterrors.ToGRPC(vtgErr)
}
545701
func (sn *VTGateSession) StreamExecute(ctx context.Context, query string, bindVars map[string]*querypb.BindVariable) (sqltypes.ResultStream, error) {
	// StreamExecute is only used for SELECT queries that don't change
	// the session. So, the protocol doesn't return an updated session.
	// This may change in the future.
	return sn.impl.StreamExecute(ctx, sn.session, query, bindVars)
}
1246974
func selectRows(row virtualRow, predicates []PredicateLinker, functors []selectFunctor) error {
	var res bool
	var err error

	// If the row validate all predicates, write it
	for _, predicate := range predicates {
		if res, err = predicate.Eval(row); err != nil {
			return err
		}
		if res == false {
			return nil
		}
	}

	for i := range functors {
		err := functors[i].FeedVirtualRow(row)
		if err != nil {
			return err
		}
	}
	return nil
}
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:]...),
	}
}
635085
func (o *rawSet) RowsToStruct(ptrStruct interface{}, keyCol, valueCol string) (int64, error) {
	return o.queryRowsTo(ptrStruct, keyCol, valueCol)
}
1156395
func (tx *Tx) QueryRowx(query string, args ...interface{}) *Row {
	rows, err := tx.Tx.Query(query, args...)
	return &Row{rows: rows, err: err, unsafe: tx.unsafe, Mapper: tx.Mapper}
}
880324
func (c *Host) StreamVolumes(ch chan *volume.Event) (stream.Stream, error) {
	return c.c.ResumingStream("GET", "/storage/volumes", ch)
}
1252995
func (q *Qbs) QueryRow(query string, args ...interface{}) *sql.Row {
	q.log(query, args...)
	query = q.Dialect.substituteMarkers(query)
	stmt, err := q.prepare(query)
	if err != nil {
		q.updateTxError(err)
		return nil
	}
	return stmt.QueryRow(args...)
}
339938
func DispatchStreamHandler(
	h StreamHandler,
	stream *ServerStream,
) (err error) {
	return InvokeStreamHandler(StreamInvokeRequest{
		Stream:  stream,
		Handler: h,
	})
}
1001221
func streamemitRows(qw422016 *qt422016.Writer, rows []string) {
	
	qw422016.N().S(`
	<tr>
		<th>#</th>
		<th>value</th>
	</tr>

	`)
	//line templates/tablepage.qtpl:35
	for n, r := range rows {
		//line templates/tablepage.qtpl:35
		qw422016.N().S(`
		`)
		//line templates/tablepage.qtpl:36
		if r == "bingo" {
			//line templates/tablepage.qtpl:36
			qw422016.N().S(`
			<tr><td colspan="2"><h1>BINGO!</h1></td></tr>
			`)
			//line templates/tablepage.qtpl:38
			return
			//line templates/tablepage.qtpl:39
		} else if n == 42 {
			//line templates/tablepage.qtpl:39
			qw422016.N().S(`
			<tr><td colspan="2">42 rows already generated</td></tr>
			`)
			//line templates/tablepage.qtpl:41
			break
			//line templates/tablepage.qtpl:42
		}
		//line templates/tablepage.qtpl:42
		qw422016.N().S(`

		<tr style="background: `)
		//line templates/tablepage.qtpl:44
		if n&1 == 1 {
			//line templates/tablepage.qtpl:44
			qw422016.N().S(`white`)
			//line templates/tablepage.qtpl:44
		} else {
			//line templates/tablepage.qtpl:44
			qw422016.N().S(`#ddd`)
			//line templates/tablepage.qtpl:44
		}
		//line templates/tablepage.qtpl:44
		qw422016.N().S(`">
			<td>`)
		//line templates/tablepage.qtpl:45
		qw422016.N().D(n + 1)
		//line templates/tablepage.qtpl:45
		qw422016.N().S(`</td>
			<td>`)
		//line templates/tablepage.qtpl:46
		qw422016.E().S(r)
		//line templates/tablepage.qtpl:46
		qw422016.N().S(`</td>
		</tr>
	`)
		//line templates/tablepage.qtpl:48
	}
	//line templates/tablepage.qtpl:48
	qw422016.N().S(`

	<tr><td colspan="2">No bingo found</td></tr>
`)
//line templates/tablepage.qtpl:51
}