func errUnsupportedMergeType(a *proto3.Value) error {
	return spannerErrorf(codes.FailedPrecondition, "unsupported type merge (%T)", a.Kind)
}
errUnsupportedMergeType returns error for protobuf type that cannot be
merged to other protobufs.
package spanner
import (
	"bytes"
	"io"
	"log"
	"sync/atomic"
	"time"
	"cloud.google.com/go/internal/protostruct"
	proto "github.com/golang/protobuf/proto"
	proto3 "github.com/golang/protobuf/ptypes/struct"
	"golang.org/x/net/context"
	"google.golang.org/api/iterator"
	sppb "google.golang.org/genproto/googleapis/spanner/v1"
	"google.golang.org/grpc/codes"
)
type streamingReceiver interface {
	Recv() (*sppb.PartialResultSet, error)
}
func errEarlyReadEnd() error {
	return spannerErrorf(codes.FailedPrecondition, "read completed with active stream")
}
func stream(ctx context.Context, rpc func(ct context.Context, resumeToken []byte) (streamingReceiver, error), setTimestamp func(time.Time), release func(error)) *RowIterator {
	ctx, cancel := context.WithCancel(ctx)
	ctx = traceStartSpan(ctx, "cloud.google.com/go/spanner.RowIterator")
	return &RowIterator{
		streamd:      newResumableStreamDecoder(ctx, rpc),
		rowd:         &partialResultSetDecoder{},
		setTimestamp: setTimestamp,
		release:      release,
		cancel:       cancel,
	}
}
type RowIterator struct {
	QueryPlan *sppb.QueryPlan
	QueryStats map[string]interface{}
	streamd      *resumableStreamDecoder
	rowd         *partialResultSetDecoder
	setTimestamp func(time.Time)
	release      func(error)
	cancel       func()
	err          error
	rows         []*Row
}
func (r *RowIterator) Next() (*Row, error) {
	if r.err != nil {
		return nil, r.err
	}
	for len(r.rows) == 0 && r.streamd.next() {
		prs := r.streamd.get()
		if prs.Stats != nil {
			r.QueryPlan = prs.Stats.QueryPlan
			r.QueryStats = protostruct.DecodeToMap(prs.Stats.QueryStats)
		}
		r.rows, r.err = r.rowd.add(prs)
		if r.err != nil {
			return nil, r.err
		}
		if !r.rowd.ts.IsZero() && r.setTimestamp != nil {
			r.setTimestamp(r.rowd.ts)
			r.setTimestamp = nil
		}
	}
	if len(r.rows) > 0 {
		row := r.rows[0]
		r.rows = r.rows[1:]
		return row, nil
	}
	if err := r.streamd.lastErr(); err != nil {
		r.err = toSpannerError(err)
	} else if !r.rowd.done() {
		r.err = errEarlyReadEnd()
	} else {
		r.err = iterator.Done
	}
	return nil, r.err
}
func (r *RowIterator) Do(f func(r *Row) error) error {
	defer r.Stop()
	for {
		row, err := r.Next()
		switch err {
		case iterator.Done:
			return nil
		case nil:
			if err = f(row); err != nil {
				return err
			}
		default:
			return err
		}
	}
}
func (r *RowIterator) Stop() {
	if r.streamd != nil {
		defer traceEndSpan(r.streamd.ctx, r.err)
	}
	if r.cancel != nil {
		r.cancel()
	}
	if r.release != nil {
		r.release(r.err)
		if r.err == nil {
			r.err = spannerErrorf(codes.FailedPrecondition, "Next called after Stop")
		}
		r.release = nil
	}
}
type partialResultQueue struct {
	q     []*sppb.PartialResultSet
	first int
	last  int
	n     int  
}
func (q *partialResultQueue) empty() bool {
	return q.n == 0
}
func errEmptyQueue() error {
	return spannerErrorf(codes.OutOfRange, "empty partialResultQueue")
}
func (q *partialResultQueue) peekLast() (*sppb.PartialResultSet, error) {
	if q.empty() {
		return nil, errEmptyQueue()
	}
	return q.q[(q.last+cap(q.q)-1)%cap(q.q)], nil
}
func (q *partialResultQueue) push(r *sppb.PartialResultSet) {
	if q.q == nil {
		q.q = make([]*sppb.PartialResultSet, 8  )
	}
	if q.n == cap(q.q) {
		buf := make([]*sppb.PartialResultSet, cap(q.q)*2)
		for i := 0; i < q.n; i++ {
			buf[i] = q.q[(q.first+i)%cap(q.q)]
		}
		q.q = buf
		q.first = 0
		q.last = q.n
	}
	q.q[q.last] = r
	q.last = (q.last + 1) % cap(q.q)
	q.n++
}
func (q *partialResultQueue) pop() *sppb.PartialResultSet {
	if q.n == 0 {
		return nil
	}
	r := q.q[q.first]
	q.q[q.first] = nil
	q.first = (q.first + 1) % cap(q.q)
	q.n--
	return r
}
func (q *partialResultQueue) clear() {
	*q = partialResultQueue{}
}
func (q *partialResultQueue) dump() []*sppb.PartialResultSet {
	var dq []*sppb.PartialResultSet
	for i := q.first; len(dq) < q.n; i = (i + 1) % cap(q.q) {
		dq = append(dq, q.q[i])
	}
	return dq
}
type resumableStreamDecoderState int
const (
	unConnected         resumableStreamDecoderState = iota  
	queueingRetryable                                       
	queueingUnretryable                                     
	aborted                                                 
	finished                                                
)
type resumableStreamDecoder struct {
	state resumableStreamDecoderState
	stateWitness func(resumableStreamDecoderState)
	ctx context.Context
	rpc func(ctx context.Context, restartToken []byte) (streamingReceiver, error)
	stream streamingReceiver
	q partialResultQueue
	bytesBetweenResumeTokens int32
	maxBytesBetweenResumeTokens int32
	np *sppb.PartialResultSet
	resumeToken []byte
	retryCount int
	err error
	backoff exponentialBackoff
}
func newResumableStreamDecoder(ctx context.Context, rpc func(ct context.Context, restartToken []byte) (streamingReceiver, error)) *resumableStreamDecoder {
	return &resumableStreamDecoder{
		ctx: ctx,
		rpc: rpc,
		maxBytesBetweenResumeTokens: atomic.LoadInt32(&maxBytesBetweenResumeTokens),
		backoff:                     defaultBackoff,
	}
}
func (d *resumableStreamDecoder) changeState(target resumableStreamDecoderState) {
	if d.state == queueingRetryable && d.state != target {
		d.bytesBetweenResumeTokens = 0
	}
	d.state = target
	if d.stateWitness != nil {
		d.stateWitness(target)
	}
}
func (d *resumableStreamDecoder) isNewResumeToken(rt []byte) bool {
	if rt == nil {
		return false
	}
	if bytes.Compare(rt, d.resumeToken) == 0 {
		return false
	}
	return true
}
var (
	maxBytesBetweenResumeTokens = int32(128 * 1024 * 1024)
)
func (d *resumableStreamDecoder) next() bool {
	for {
		select {
		case <-d.ctx.Done():
			d.err = errContextCanceled(d.ctx, d.err)
			d.changeState(aborted)
		default:
		}
		switch d.state {
		case unConnected:
			if d.stream, d.err = d.rpc(d.ctx, d.resumeToken); d.err != nil {
				if isRetryable(d.err) {
					d.doBackOff()
					d.changeState(unConnected)
					continue
				}
				d.changeState(aborted)
				continue
			}
			d.resetBackOff()
			d.changeState(queueingRetryable)
			continue
		case queueingRetryable:
			fallthrough
		case queueingUnretryable:
			last, err := d.q.peekLast()
			if err != nil {
				d.tryRecv()
				continue
			}
			if d.isNewResumeToken(last.ResumeToken) {
				d.np = d.q.pop()
				if d.q.empty() {
					d.bytesBetweenResumeTokens = 0
					d.resumeToken = d.np.ResumeToken
					d.changeState(queueingRetryable)
				}
				return true
			}
			if d.bytesBetweenResumeTokens >= d.maxBytesBetweenResumeTokens && d.state == queueingRetryable {
				d.changeState(queueingUnretryable)
				continue
			}
			if d.state == queueingUnretryable {
				d.np = d.q.pop()
				return true
			}
			d.tryRecv()
			continue
		case aborted:
			d.q.clear()
			return false
		case finished:
			if d.q.empty() {
				return false
			}
			d.np = d.q.pop()
			return true
		default:
			log.Printf("Unexpected resumableStreamDecoder.state: %v", d.state)
			return false
		}
	}
}
func (d *resumableStreamDecoder) tryRecv() {
	var res *sppb.PartialResultSet
	if res, d.err = d.stream.Recv(); d.err != nil {
		if d.err == io.EOF {
			d.err = nil
			d.changeState(finished)
			return
		}
		if isRetryable(d.err) && d.state == queueingRetryable {
			d.err = nil
			d.q.clear()
			d.stream = nil
			d.changeState(unConnected)
			d.doBackOff()
			return
		}
		d.changeState(aborted)
		return
	}
	d.q.push(res)
	if d.state == queueingRetryable && !d.isNewResumeToken(res.ResumeToken) {
		d.bytesBetweenResumeTokens += int32(proto.Size(res))
	}
	d.resetBackOff()
	d.changeState(d.state)
}
func (d *resumableStreamDecoder) resetBackOff() {
	d.retryCount = 0
}
func (d *resumableStreamDecoder) doBackOff() {
	delay := d.backoff.delay(d.retryCount)
	tracePrintf(d.ctx, nil, "Backing off stream read for %s", delay)
	ticker := time.NewTicker(delay)
	defer ticker.Stop()
	d.retryCount++
	select {
	case <-d.ctx.Done():
	case <-ticker.C:
	}
}
func (d *resumableStreamDecoder) get() *sppb.PartialResultSet {
	return d.np
}
func (d *resumableStreamDecoder) lastErr() error {
	return d.err
}
type partialResultSetDecoder struct {
	row     Row
	tx      *sppb.Transaction
	chunked bool       
	ts      time.Time  
}
func (p *partialResultSetDecoder) yield(chunked, last bool) *Row {
	if len(p.row.vals) == len(p.row.fields) && (!chunked || !last) {
		fresh := Row{
			fields: p.row.fields,
			vals:   make([]*proto3.Value, len(p.row.vals)),
		}
		copy(fresh.vals, p.row.vals)
		p.row.vals = p.row.vals[:0]  
		return &fresh
	}
	return nil
}
func errChunkedEmptyRow() error {
	return spannerErrorf(codes.FailedPrecondition, "got invalid chunked PartialResultSet with empty Row")
}
func (p *partialResultSetDecoder) add(r *sppb.PartialResultSet) ([]*Row, error) {
	var rows []*Row
	if r.Metadata != nil {
		if p.row.fields == nil {
			p.row.fields = r.Metadata.RowType.Fields
		}
		if p.tx == nil && r.Metadata.Transaction != nil {
			p.tx = r.Metadata.Transaction
			if p.tx.ReadTimestamp != nil {
				p.ts = time.Unix(p.tx.ReadTimestamp.Seconds, int64(p.tx.ReadTimestamp.Nanos))
			}
		}
	}
	if len(r.Values) == 0 {
		return nil, nil
	}
	if p.chunked {
		p.chunked = false
		last := len(p.row.vals) - 1
		if last < 0 {  
			return nil, errChunkedEmptyRow()
		}
		var err error
		if p.row.vals[last], err = p.merge(p.row.vals[last], r.Values[0]); err != nil {
			return nil, err
		}
		r.Values = r.Values[1:]
		if row := p.yield(r.ChunkedValue, len(r.Values) == 0); row != nil {
			rows = append(rows, row)
		}
	}
	for i, v := range r.Values {
		p.row.vals = append(p.row.vals, v)
		if row := p.yield(r.ChunkedValue, i == len(r.Values)-1); row != nil {
			rows = append(rows, row)
		}
	}
	if r.ChunkedValue {
		p.chunked = true
	}
	return rows, nil
}
func (p *partialResultSetDecoder) isMergeable(a *proto3.Value) bool {
	switch a.Kind.(type) {
	case *proto3.Value_StringValue:
		return true
	case *proto3.Value_ListValue:
		return true
	default:
		return false
	}
}
func errIncompatibleMergeTypes(a, b *proto3.Value) error {
	return spannerErrorf(codes.FailedPrecondition, "incompatible type in chunked PartialResultSet. expected (%T), got (%T)", a.Kind, b.Kind)
}