func (n *NDTParser) TableName() string {
	return n.TableBase()
}
TableName returns the base of the bq table inserter target.
package parser
import (
	"errors"
	"fmt"
	"log"
	"os"
	"reflect"
	"regexp"
	"strconv"
	"strings"
	"time"
	"github.com/m-lab/go/logx"
	"cloud.google.com/go/bigquery"
	"github.com/m-lab/annotation-service/api"
	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/schema"
	"github.com/m-lab/etl/web100"
	"github.com/prometheus/client_golang/prometheus"
)
var (
	NDTEstimateBW, _ = strconv.ParseBool(os.Getenv("NDT_ESTIMATE_BW"))
)
const (
	minNumSnapshots = 1600  
	maxNumSnapshots = 2800  
)
const dateDir = `^(?P<dir>\d{4}/\d{2}/\d{2}/)?`
const dateField = `(?P<date>\d{8})`
const timeField = `(?P<time>[012]\d:[0-6]\d:\d{2}\.\d{1,10})`
const address = `(?P<address>.*)`
const suffix = `(?P<suffix>[a-z2].*)`
var (
	testFilePattern = regexp.MustCompile(
		"^" + dateDir + dateField + "T" + timeField + "Z_" + address + `\.` + suffix + "$")
	gzTestFilePattern = regexp.MustCompile(
		"^" + dateDir + dateField + "T" + timeField + "Z_" + address + `\.` + suffix + `\.gz$`)
	datePattern = regexp.MustCompile(dateField)
	timePattern = regexp.MustCompile("T" + timeField + "Z_")
	endPattern  = regexp.MustCompile(suffix + `$`)
)
type TestInfo struct {
	DateDir   string     
	Date      string     
	Time      string     
	Address   string     
	Suffix    string     
	Timestamp time.Time  
}
func ParseNDTFileName(path string) (*TestInfo, error) {
	fields := gzTestFilePattern.FindStringSubmatch(path)
	if fields == nil {
		fields = testFilePattern.FindStringSubmatch(path)
	}
	if fields == nil {
		if !datePattern.MatchString(path) {
			return nil, errors.New("Path should contain yyyymmddT: " + path)
		} else if !timePattern.MatchString(path) {
			return nil, errors.New("Path should contain Thh:mm:ss.ff...Z_: " + path)
		} else if !endPattern.MatchString(path) {
			return nil, errors.New("Path should end in \\.[a-z2].*: " + path)
		}
		return nil, errors.New("Invalid test path: " + path)
	}
	timestamp, err := time.Parse("20060102T15:04:05.999999999Z_", fields[2]+"T"+fields[3]+"Z_")
	if err != nil {
		log.Println(fields[2] + "T" + fields[3] + "   " + err.Error())
		return nil, errors.New("Invalid test path: " + path)
	}
	return &TestInfo{fields[1], fields[2], fields[3], fields[4], fields[5], timestamp}, nil
}
type fileInfoAndData struct {
	fn   string
	info TestInfo
	data []byte
}
type NDTParser struct {
	Base
	taskFileName string  
	timestamp    string  
	c2s *fileInfoAndData
	s2c *fileInfoAndData
	metaFile *MetaFileData
}
func NewNDTParser(ins etl.Inserter, annotator ...v2as.Annotator) *NDTParser {
	bufSize := etl.NDT.BQBufferSize()
	var ann v2as.Annotator
	if len(annotator) > 0 && annotator[0] != nil {
		ann = annotator[0]
	} else {
		ann = v2as.GetAnnotator(etl.BatchAnnotatorURL)
	}
	return &NDTParser{Base: *NewBase(ins, bufSize, ann)}
}
func (n *NDTParser) TaskError() error {
	if n.Inserter.Committed() < 10*n.Inserter.Failed() {
		log.Printf("Warning: high row commit errors: %d / %d\n",
			n.Inserter.Accepted(), n.Inserter.Failed())
		return errors.New("too many insertion failures")
	}
	return nil
}
func (n *NDTParser) Flush() error {
	if n.timestamp != "" {
		n.processGroup()
	}
	return n.Base.AnnotateAndFlush(n.TableBase())
}