<commit_before>/*
Copyright 2017 Google Inc. All Rights Reserved.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

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"
)

// streamingReceiver is the interface for receiving data from a client side
// stream.
type streamingReceiver interface {
	Recv() (*sppb.PartialResultSet, error)
}

// errEarlyReadEnd returns error for read finishes when gRPC stream is still active.
func errEarlyReadEnd() error {
	return spannerErrorf(codes.FailedPrecondition, "read completed with active stream")
}

// stream is the internal fault tolerant method for streaming data from
// Cloud Spanner.
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,
	}
}

// RowIterator is an iterator over Rows.
type RowIterator struct {
	// The plan for the query. Available after RowIterator.Next returns iterator.Done
	// if QueryWithStats was called.
	QueryPlan *sppb.QueryPlan

	// Execution statistics for the query. Available after RowIterator.Next returns iterator.Done
	// if QueryWithStats was called.
	QueryStats map[string]interface{}

	streamd      *resumableStreamDecoder
	rowd         *partialResultSetDecoder
	setTimestamp func(time.Time)
	release      func(error)
	cancel       func()
	err          error
	rows         []*Row
}

// Next returns the next result. Its second return value is iterator.Done if
// there are no more results. Once Next returns Done, all subsequent calls
// will return Done.
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
}

// Do calls the provided function once in sequence for each row in the iteration.  If the
// function returns a non-nil error, Do immediately returns that error.
//
// If there are no rows in the iterator, Do will return nil without calling the
// provided function.
//
// Do always calls Stop on the iterator.
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
		}
	}
}

// Stop terminates the iteration. It should be called after every iteration.
func (r *RowIterator) Stop() {
	if r.streamd != nil {
		defer traceEndSpan(r.streamd.ctx, r.err)
	}<commit_msg>spanner: clarify RowIterator.Stop doc<commit_after>/*
Copyright 2017 Google Inc. All Rights Reserved.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

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"
)

// streamingReceiver is the interface for receiving data from a client side
// stream.
type streamingReceiver interface {
	Recv() (*sppb.PartialResultSet, error)
}

// errEarlyReadEnd returns error for read finishes when gRPC stream is still active.
func errEarlyReadEnd() error {
	return spannerErrorf(codes.FailedPrecondition, "read completed with active stream")
}

// stream is the internal fault tolerant method for streaming data from
// Cloud Spanner.
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,
	}
}

// RowIterator is an iterator over Rows.
type RowIterator struct {
	// The plan for the query. Available after RowIterator.Next returns iterator.Done
	// if QueryWithStats was called.
	QueryPlan *sppb.QueryPlan

	// Execution statistics for the query. Available after RowIterator.Next returns iterator.Done
	// if QueryWithStats was called.
	QueryStats map[string]interface{}

	streamd      *resumableStreamDecoder
	rowd         *partialResultSetDecoder
	setTimestamp func(time.Time)
	release      func(error)
	cancel       func()
	err          error
	rows         []*Row
}

// Next returns the next result. Its second return value is iterator.Done if
// there are no more results. Once Next returns Done, all subsequent calls
// will return Done.
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
}

// Do calls the provided function once in sequence for each row in the iteration.  If the
// function returns a non-nil error, Do immediately returns that error.
//
// If there are no rows in the iterator, Do will return nil without calling the
// provided function.
//
// Do always calls Stop on the iterator.
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
		}
	}
}

// Stop terminates the iteration. It should be called after you finish using the iterator.
func (r *RowIterator) Stop() {
	if r.streamd != nil {
		defer traceEndSpan(r.streamd.ctx, r.err)
	}
{'fraction_non_alphanumeric': 0.10778443113772455, 'fraction_numerical': 0.0033532934131736527, 'mean_word_length': 3.4849624060150375, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6563208', 'n_tokens_mistral': 2813, 'n_tokens_neox': 2646, 'n_words': 896}