func (dp *NDT5ResultParser) RowsInBuffer() int {
	return dp.GetStats().Pending
}
RowsInBuffer returns the count of rows currently in the buffer.
package parser
import (
	"bytes"
	"encoding/json"
	"log"
	"regexp"
	"strings"
	"time"
	"cloud.google.com/go/bigquery"
	v2as "github.com/m-lab/annotation-service/api/v2"
	"github.com/m-lab/etl/etl"
	"github.com/m-lab/etl/metrics"
	"github.com/m-lab/etl/row"
	"github.com/m-lab/etl/schema"
)
type NDT5ResultParser struct {
	*row.Base
	table  string
	suffix string
}
func NewNDT5ResultParser(sink row.Sink, label, suffix string, ann v2as.Annotator) etl.Parser {
	bufSize := etl.NDT5.BQBufferSize()
	if ann == nil {
		ann = v2as.GetAnnotator(etl.BatchAnnotatorURL)
	}
	return &NDT5ResultParser{
		Base:   row.NewBase(label, sink, bufSize, ann),
		table:  label,
		suffix: suffix,
	}
}
func (dp *NDT5ResultParser) TaskError() error {
	stats := dp.GetStats()
	if stats.Total() < 10*stats.Failed {
		log.Printf("Warning: high row commit errors (more than 10%%): %d failed of %d accepted\n",
			stats.Failed, stats.Total())
		return etl.ErrHighInsertionFailureRate
	}
	return nil
}
func (dp *NDT5ResultParser) IsParsable(testName string, data []byte) (string, bool) {
	if strings.HasSuffix(testName, "json") {
		return "ndt5_result", true
	}
	return "unknown", false
}
func (dp *NDT5ResultParser) ParseAndInsert(meta map[string]bigquery.Value, testName string, test []byte) error {
	metrics.WorkerState.WithLabelValues(dp.TableName(), "ndt5_result").Inc()
	defer metrics.WorkerState.WithLabelValues(dp.TableName(), "ndt5_result").Dec()
	var re = regexp.MustCompile(`,"ClientMetadata":{[^}]+}`)
	test = []byte(re.ReplaceAllString(string(test), ``))
	rdr := bytes.NewReader(test)
	dec := json.NewDecoder(rdr)
	for dec.More() {
		stats := schema.NDT5ResultRow{
			TestID: testName,
			ParseInfo: &schema.ParseInfoV0{
				TaskFileName:  meta["filename"].(string),
				ParseTime:     time.Now(),
				ParserVersion: Version(),
			},
		}
		err := dec.Decode(&stats.Result)
		if err != nil {
			log.Println(err)
			metrics.TestCount.WithLabelValues(
				dp.TableName(), "ndt5_result", "Decode").Inc()
			return err
		}
		stats.LogTime = stats.Result.StartTime.Unix()
		metrics.RowSizeHistogram.WithLabelValues(
			dp.TableName()).Observe(float64(len(test)))
		if err = dp.Base.Put(&stats); err != nil {
			return err
		}
		metrics.TestCount.WithLabelValues(dp.TableName(), "ndt5_result", "ok").Inc()
	}
	return nil
}
func (dp *NDT5ResultParser) Flush() error {
	return dp.Base.Flush()
}
func (dp *NDT5ResultParser) TableName() string {
	return dp.table
}
func (dp *NDT5ResultParser) FullTableName() string {
	return dp.table + dp.suffix
}