text:
package executor

import (
	"bytes"
	"math"
	"sort"
	"sync"
	"time"

	"github.com/cznic/sortutil"
	"github.com/juju/errors"
	"github.com/pingcap/tidb/ast"
	"github.com/pingcap/tidb/context"
	"github.com/pingcap/tidb/distsql"
	"github.com/pingcap/tidb/domain"
	"github.com/pingcap/tidb/expression"
	"github.com/pingcap/tidb/infoschema"
	"github.com/pingcap/tidb/kv"
	"github.com/pingcap/tidb/model"
	"github.com/pingcap/tidb/plan"
	"github.com/pingcap/tidb/sessionctx/stmtctx"
	"github.com/pingcap/tidb/statistics"
	"github.com/pingcap/tidb/table"
	"github.com/pingcap/tidb/types"
	"github.com/pingcap/tidb/util/admin"
	"github.com/pingcap/tidb/util/chunk"
	"github.com/pingcap/tidb/util/ranger"
	"github.com/pingcap/tipb/go-tipb"
	goctx "golang.org/x/net/context"
)

// executorBuilder builds an Executor from a Plan.
// The InfoSchema must not change during execution.
type executorBuilder struct {
	ctx      context.Context
	is       infoschema.InfoSchema
	priority int
	startTS  uint64 // cached when the first time getStartTS() is called
	// err is set when there is error happened during Executor building process.
	err error
}

func newExecutorBuilder(ctx context.Context, is infoschema.InfoSchema, priority int) *executorBuilder {
	return &executorBuilder{
		ctx:      ctx,
		is:       is,
		priority: priority,
	}
}

func (b *executorBuilder) build(p plan.Plan) Executor {
	switch v := p.(type) {
	case nil:
		return nil
	case *plan.CheckTable:
		return b.buildCheckTable(v)
	case *plan.DDL:
		return b.buildDDL(v)
	case *plan.Deallocate:
		return b.buildDeallocate(v)
	case *plan.Delete:
		return b.buildDelete(v)
	case *plan.Execute:
		return b.buildExecute(v)
	case *plan.Explain:
		return b.buildExplain(v)
	case *plan.Insert:
		return b.buildInsert(v)
	case *plan.LoadData:
		return b.buildLoadData(v)
	case *plan.PhysicalLimit:
		return b.buildLimit(v)
	case *plan.Prepare:
		return b.buildPrepare(v)
	case *plan.PhysicalLock:
		return b.buildSelectLock(v)
	case *plan.CancelDDLJobs:
		return b.buildCancelDDLJobs(v)
	case *plan.ShowDDL:
		return b.buildShowDDL(v)
	case *plan.ShowDDLJobs:
		return b.buildShowDDLJobs(v)
	case *plan.Show:
		return b.buildShow(v)
	case *plan.Simple:
		return b.buildSimple(v)
	case *plan.Set:
		return b.buildSet(v)
	case *plan.PhysicalSort:
		return b.buildSort(v)
	case *plan.PhysicalTopN:
		return b.buildTopN(v)
	case *plan.PhysicalUnionAll:
		return b.buildUnionAll(v)
	case *plan.Update:
		return b.buildUpdate(v)
	case *plan.PhysicalUnionScan:
		return b.buildUnionScanExec(v)
	case *plan.PhysicalHashJoin:
		return b.buildHashJoin(v)
	case *plan.PhysicalMergeJoin:
		return b.buildMergeJoin(v)
	case *plan.PhysicalIndexJoin:
		return b.buildIndexLookUpJoin(v)
	case *plan.PhysicalSelection:
		return b.buildSelection(v)
	case *plan.PhysicalHashAgg:
		return b.buildHashAgg(v)
	case *plan.PhysicalStreamAgg:
		return b.buildStreamAgg(v)
	case *plan.PhysicalProjection:
		return b.buildProjection(v)
	case *plan.PhysicalMemTable:
		return b.buildMemTable(v)
	case *plan.PhysicalTableDual:
		return b.buildTableDual(v)
	case *plan.PhysicalApply:
		return b.buildApply(v)
	case *plan.PhysicalExists:
		return b.buildExists(v)
	case *plan.PhysicalMaxOneRow:
		return b.buildMaxOneRow(v)
	case *plan.Analyze:
		return b.buildAnalyze(v)
	case *plan.PhysicalTableReader:
		return b.buildTableReader(v)
	case *plan.PhysicalIndexReader:
		return b.buildIndexReader(v)
	case *plan.PhysicalIndexLookUpReader:
		return b.buildIndexLookUpReader(v)
	default:
		b.err = ErrUnknownPlan.Gen("Unknown Plan %T", p)
		return nil
	}
}

func (b *executorBuilder) buildCancelDDLJobs(v *plan.CancelDDLJobs) Executor {
	e := &CancelDDLJobsExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
		jobIDs:       v.JobIDs,
	}
	e.errs, b.err = admin.CancelJobs(e.ctx.Txn(), e.jobIDs)
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildShowDDL(v *plan.ShowDDL) Executor {
	// We get DDLInfo here because for Executors that returns result set,
	// next will be called after transaction has been committed.
	// We need the transaction to get DDLInfo.
	e := &ShowDDLExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
	}

	var err error
	ownerManager := domain.GetDomain(e.ctx).DDL().OwnerManager()
	ctx, cancel := goctx.WithTimeout(goctx.Background(), 3*time.Second)
	e.ddlOwnerID, err = ownerManager.GetOwnerID(ctx)
	cancel()
	if err != nil {
		b.err = errors.Trace(err)
		return nil
	}

	ddlInfo, err := admin.GetDDLInfo(e.ctx.Txn())
	if err != nil {
		b.err = errors.Trace(err)
		return nil
	}
	e.ddlInfo = ddlInfo
	e.selfID = ownerManager.ID()
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildShowDDLJobs(v *plan.ShowDDLJobs) Executor {
	e := &ShowDDLJobsExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildCheckTable(v *plan.CheckTable) Executor {
	e := &CheckTableExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
		tables:       v.Tables,
		is:           b.is,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildDeallocate(v *plan.Deallocate) Executor {
	e := &DeallocateExec{
		baseExecutor: newBaseExecutor(nil, b.ctx),
		Name:         v.Name,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildSelectLock(v *plan.PhysicalLock) Executor {
	src := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	if !b.ctx.GetSessionVars().InTxn() {
		// Locking of rows for update using SELECT FOR UPDATE only applies when autocommit
		// is disabled (either by beginning transaction with START TRANSACTION or by setting
		// autocommit to 0. If autocommit is enabled, the rows matching the specification are not locked.
		// See https://dev.mysql.com/doc/refman/5.7/en/innodb-locking-reads.html
		return src
	}
	e := &SelectLockExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, src),
		Lock:         v.Lock,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildLimit(v *plan.PhysicalLimit) Executor {
	childExec := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	e := &LimitExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, childExec),
		begin:        v.Offset,
		end:          v.Offset + v.Count,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildPrepare(v *plan.Prepare) Executor {
	e := &PrepareExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
		is:           b.is,
		name:         v.Name,
		sqlText:      v.SQLText,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildExecute(v *plan.Execute) Executor {
	return &ExecuteExec{
		baseExecutor: newBaseExecutor(nil, b.ctx),
		IS:           b.is,
		Name:         v.Name,
		UsingVars:    v.UsingVars,
		ID:           v.ExecID,
		Stmt:         v.Stmt,
		Plan:         v.Plan,
	}
}

func (b *executorBuilder) buildShow(v *plan.Show) Executor {
	e := &ShowExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
		Tp:           v.Tp,
		DBName:       model.NewCIStr(v.DBName),
		Table:        v.Table,
		Column:       v.Column,
		User:         v.User,
		Flag:         v.Flag,
		Full:         v.Full,
		GlobalScope:  v.GlobalScope,
		is:           b.is,
	}
	e.supportChk = true
	if e.Tp == ast.ShowGrants && e.User == nil {
		e.User = e.ctx.GetSessionVars().User
	}
	if len(v.Conditions) == 0 {
		return e
	}
	sel := &SelectionExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, e),
		filters:      v.Conditions,
	}
	sel.supportChk = true
	return sel
}

func (b *executorBuilder) buildSimple(v *plan.Simple) Executor {
	switch s := v.Statement.(type) {
	case *ast.GrantStmt:
		return b.buildGrant(s)
	case *ast.RevokeStmt:
		return b.buildRevoke(s)
	}
	e := &SimpleExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
		Statement:    v.Statement,
		is:           b.is,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildSet(v *plan.Set) Executor {
	e := &SetExecutor{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
		vars:         v.VarAssigns,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildInsert(v *plan.Insert) Executor {
	selectExec := b.build(v.SelectPlan)
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	var baseExec baseExecutor
	if selectExec != nil {
		baseExec = newBaseExecutor(nil, b.ctx, selectExec)
	} else {
		baseExec = newBaseExecutor(nil, b.ctx)
	}

	ivs := &InsertValues{
		baseExecutor:          baseExec,
		Columns:               v.Columns,
		Lists:                 v.Lists,
		Setlist:               v.Setlist,
		GenColumns:            v.GenCols.Columns,
		GenExprs:              v.GenCols.Exprs,
		needFillDefaultValues: v.NeedFillDefaultValue,
		SelectExec:            selectExec,
	}

	ivs.Table = v.Table
	ivs.initDefaultValBuf()
	if v.IsReplace {
		return b.buildReplace(ivs)
	}
	insert := &InsertExec{
		InsertValues: ivs,
		OnDuplicate:  append(v.OnDuplicate, v.GenCols.OnDuplicates...),
		Priority:     v.Priority,
		IgnoreErr:    v.IgnoreErr,
	}
	insert.supportChk = true
	return insert
}

func (b *executorBuilder) buildLoadData(v *plan.LoadData) Executor {
	tbl, ok := b.is.TableByID(v.Table.TableInfo.ID)
	if !ok {
		b.err = errors.Errorf("Can not get table %d", v.Table.TableInfo.ID)
		return nil
	}
	insertVal := &InsertValues{
		baseExecutor: newBaseExecutor(nil, b.ctx),
		Table:        tbl,
		Columns:      v.Columns,
		GenColumns:   v.GenCols.Columns,
		GenExprs:     v.GenCols.Exprs,
	}
	insertVal.initDefaultValBuf()
	tableCols := tbl.Cols()
	columns, err := insertVal.getColumns(tableCols)
	if err != nil {
		b.err = errors.Trace(err)
		return nil
	}
	loadDataExec := &LoadData{
		baseExecutor: newBaseExecutor(nil, b.ctx),
		IsLocal:      v.IsLocal,
		loadDataInfo: &LoadDataInfo{
			row:        make([]types.Datum, len(columns)),
			insertVal:  insertVal,
			Path:       v.Path,
			Table:      tbl,
			FieldsInfo: v.FieldsInfo,
			LinesInfo:  v.LinesInfo,
			Ctx:        b.ctx,
			columns:    columns,
		},
	}

	loadDataExec.supportChk = true
	return loadDataExec
}

func (b *executorBuilder) buildReplace(vals *InsertValues) Executor {
	replaceExec := &ReplaceExec{
		InsertValues: vals,
	}
	replaceExec.supportChk = true
	return replaceExec
}

func (b *executorBuilder) buildGrant(grant *ast.GrantStmt) Executor {
	e := &GrantExec{
		baseExecutor: newBaseExecutor(nil, b.ctx),
		Privs:        grant.Privs,
		ObjectType:   grant.ObjectType,
		Level:        grant.Level,
		Users:        grant.Users,
		WithGrant:    grant.WithGrant,
		is:           b.is,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildRevoke(revoke *ast.RevokeStmt) Executor {
	e := &RevokeExec{
		ctx:        b.ctx,
		Privs:      revoke.Privs,
		ObjectType: revoke.ObjectType,
		Level:      revoke.Level,
		Users:      revoke.Users,
		is:         b.is,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildDDL(v *plan.DDL) Executor {
	e := &DDLExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
		stmt:         v.Statement,
		is:           b.is,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildExplain(v *plan.Explain) Executor {
	e := &ExplainExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
	}
	e.rows = make([][]string, 0, len(v.Rows))
	for _, row := range v.Rows {
		e.rows = append(e.rows, row)
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildUnionScanExec(v *plan.PhysicalUnionScan) Executor {
	src := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	us := &UnionScanExec{baseExecutor: newBaseExecutor(v.Schema(), b.ctx, src)}
	// Get the handle column index of the below plan.
	// We can guarantee that there must be only one col in the map.
	for _, cols := range v.Children()[0].Schema().TblID2Handle {
		us.belowHandleIndex = cols[0].Index
	}
	switch x := src.(type) {
	case *TableReaderExecutor:
		us.desc = x.desc
		us.dirty = getDirtyDB(b.ctx).getDirtyTable(x.table.Meta().ID)
		us.conditions = v.Conditions
		us.columns = x.columns
		b.err = us.buildAndSortAddedRows()
	case *IndexReaderExecutor:
		us.desc = x.desc
		for _, ic := range x.index.Columns {
			for i, col := range x.schema.Columns {
				if col.ColName.L == ic.Name.L {
					us.usedIndex = append(us.usedIndex, i)
					break
				}
			}
		}
		us.dirty = getDirtyDB(b.ctx).getDirtyTable(x.table.Meta().ID)
		us.conditions = v.Conditions
		us.columns = x.columns
		b.err = us.buildAndSortAddedRows()
	case *IndexLookUpExecutor:
		us.desc = x.desc
		for _, ic := range x.index.Columns {
			for i, col := range x.schema.Columns {
				if col.ColName.L == ic.Name.L {
					us.usedIndex = append(us.usedIndex, i)
					break
				}
			}
		}
		us.dirty = getDirtyDB(b.ctx).getDirtyTable(x.table.Meta().ID)
		us.conditions = v.Conditions
		us.columns = x.columns
		b.err = us.buildAndSortAddedRows()
	default:
		// The mem table will not be written by sql directly, so we can omit the union scan to avoid err reporting.
		return src
	}
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	us.supportChk = true
	return us
}

// buildMergeJoin builds MergeJoinExec executor.
func (b *executorBuilder) buildMergeJoin(v *plan.PhysicalMergeJoin) Executor {
	leftExec := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}

	rightExec := b.build(v.Children()[1])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}

	leftKeys := make([]*expression.Column, 0, len(v.EqualConditions))
	rightKeys := make([]*expression.Column, 0, len(v.EqualConditions))
	for _, eqCond := range v.EqualConditions {
		if len(eqCond.GetArgs()) != 2 {
			b.err = errors.Annotate(ErrBuildExecutor, "invalid join key for equal condition")
			return nil
		}
		leftKey, ok := eqCond.GetArgs()[0].(*expression.Column)
		if !ok {
			b.err = errors.Annotate(ErrBuildExecutor, "left side of join key must be column for merge join")
			return nil
		}
		rightKey, ok := eqCond.GetArgs()[1].(*expression.Column)
		if !ok {
			b.err = errors.Annotate(ErrBuildExecutor, "right side of join key must be column for merge join")
			return nil
		}
		leftKeys = append(leftKeys, leftKey)
		rightKeys = append(rightKeys, rightKey)
	}

	lhsIter := &readerIterator{
		ctx:      b.ctx,
		reader:   leftExec,
		filter:   v.LeftConditions,
		joinKeys: leftKeys,
	}

	rhsIter := &readerIterator{
		ctx:      b.ctx,
		reader:   rightExec,
		filter:   v.RightConditions,
		joinKeys: rightKeys,
	}

	defaultValues := v.DefaultValues
	if defaultValues == nil {
		defaultValues = make([]types.Datum, rightExec.Schema().Len())
	}
	lhsColTypes := leftExec.Schema().GetTypes()
	rhsColTypes := rightExec.Schema().GetTypes()
	e := &MergeJoinExec{
		baseExecutor:    newBaseExecutor(v.Schema(), b.ctx, leftExec, rightExec),
		resultGenerator: newJoinResultGenerator(b.ctx, v.JoinType, false, defaultValues, v.OtherConditions, lhsColTypes, rhsColTypes),
		stmtCtx:         b.ctx.GetSessionVars().StmtCtx,
		// left is the outer side by default.
		outerIdx:  0,
		outerKeys: leftKeys,
		innerKeys: rightKeys,
		outerIter: lhsIter,
		innerIter: rhsIter,
	}

	if v.JoinType == plan.RightOuterJoin {
		e.outerKeys, e.innerKeys = e.innerKeys, e.outerKeys
		e.outerIter, e.innerIter = e.innerIter, e.outerIter
		e.outerIdx = 1
	}

	if v.JoinType != plan.InnerJoin {
		e.outerFilter = e.outerIter.filter
		e.outerIter.filter = nil
	}

	e.supportChk = true
	return e
}

func (b *executorBuilder) buildHashJoin(v *plan.PhysicalHashJoin) Executor {
	leftHashKey := make([]*expression.Column, 0, len(v.EqualConditions))
	rightHashKey := make([]*expression.Column, 0, len(v.EqualConditions))
	for _, eqCond := range v.EqualConditions {
		ln, _ := eqCond.GetArgs()[0].(*expression.Column)
		rn, _ := eqCond.GetArgs()[1].(*expression.Column)
		leftHashKey = append(leftHashKey, ln)
		rightHashKey = append(rightHashKey, rn)
	}

	leftExec := b.build(v.Children()[0])
	rightExec := b.build(v.Children()[1])

	// for hash join, inner table is always the smaller one.
	e := &HashJoinExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, leftExec, rightExec),
		concurrency:  v.Concurrency,
		joinType:     v.JoinType,
	}

	defaultValues := v.DefaultValues
	if v.SmallChildIdx == 0 {
		e.innerExec = leftExec
		e.outerExec = rightExec
		e.innerFilter = v.LeftConditions
		e.outerFilter = v.RightConditions
		e.innerKeys = leftHashKey
		e.outerKeys = rightHashKey
		if defaultValues == nil {
			defaultValues = make([]types.Datum, e.innerExec.Schema().Len())
		}
		e.resultGenerator = newJoinResultGenerator(b.ctx, v.JoinType, v.SmallChildIdx == 0, defaultValues,
			v.OtherConditions, nil, nil)
	} else {
		e.innerExec = rightExec
		e.outerExec = leftExec
		e.innerFilter = v.RightConditions
		e.outerFilter = v.LeftConditions
		e.innerKeys = rightHashKey
		e.outerKeys = leftHashKey
		if defaultValues == nil {
			defaultValues = make([]types.Datum, e.innerExec.Schema().Len())
		}
		e.resultGenerator = newJoinResultGenerator(b.ctx, v.JoinType, v.SmallChildIdx == 0,
			defaultValues, v.OtherConditions, nil, nil)
	}

	return e
}

func (b *executorBuilder) buildHashAgg(v *plan.PhysicalHashAgg) Executor {
	src := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	e := &HashAggExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, src),
		sc:           b.ctx.GetSessionVars().StmtCtx,
		AggFuncs:     v.AggFuncs,
		GroupByItems: v.GroupByItems,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildStreamAgg(v *plan.PhysicalStreamAgg) Executor {
	src := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	e := &StreamAggExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, src),
		StmtCtx:      b.ctx.GetSessionVars().StmtCtx,
		AggFuncs:     v.AggFuncs,
		GroupByItems: v.GroupByItems,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildSelection(v *plan.PhysicalSelection) Executor {
	childExec := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	e := &SelectionExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, childExec),
		filters:      v.Conditions,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildProjection(v *plan.PhysicalProjection) Executor {
	childExec := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	e := &ProjectionExec{
		baseExecutor:     newBaseExecutor(v.Schema(), b.ctx, childExec),
		exprs:            v.Exprs,
		evaluatorSuit:    expression.NewEvaluatorSuit(v.Exprs),
		calculateNoDelay: v.CalculateNoDelay,
	}
	e.baseExecutor.supportChk = true
	return e
}

func (b *executorBuilder) buildTableDual(v *plan.PhysicalTableDual) Executor {
	if v.RowCount != 0 && v.RowCount != 1 {
		b.err = errors.Errorf("buildTableDual failed, invalid row count for dual table: %v", v.RowCount)
		return nil
	}
	e := &TableDualExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
		numDualRows:  v.RowCount,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) getStartTS() uint64 {
	if b.startTS != 0 {
		// Return the cached value.
		return b.startTS
	}

	startTS := b.ctx.GetSessionVars().SnapshotTS
	if startTS == 0 {
		startTS = b.ctx.Txn().StartTS()
	}
	b.startTS = startTS
	return startTS
}

func (b *executorBuilder) buildMemTable(v *plan.PhysicalMemTable) Executor {
	tb, _ := b.is.TableByID(v.Table.ID)
	e := &TableScanExec{
		baseExecutor:   newBaseExecutor(v.Schema(), b.ctx),
		t:              tb,
		columns:        v.Columns,
		seekHandle:     math.MinInt64,
		ranges:         v.Ranges,
		isVirtualTable: tb.Type() == table.VirtualTable,
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildSort(v *plan.PhysicalSort) Executor {
	childExec := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	sortExec := SortExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, childExec),
		ByItems:      v.ByItems,
		schema:       v.Schema(),
	}
	sortExec.supportChk = true
	return &sortExec
}

func (b *executorBuilder) buildTopN(v *plan.PhysicalTopN) Executor {
	childExec := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	sortExec := SortExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, childExec),
		ByItems:      v.ByItems,
		schema:       v.Schema(),
	}
	sortExec.supportChk = true
	return &TopNExec{
		SortExec: sortExec,
		limit:    &plan.PhysicalLimit{Count: v.Count, Offset: v.Offset},
	}
}

func (b *executorBuilder) buildApply(apply *plan.PhysicalApply) *NestedLoopApplyExec {
	v := apply.PhysicalJoin
	leftChild := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	rightChild := b.build(v.Children()[1])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	joinSchema := expression.MergeSchema(leftChild.Schema(), rightChild.Schema())
	for _, cond := range v.EqualConditions {
		col0 := cond.GetArgs()[0].(*expression.Column)
		col0.ResolveIndices(joinSchema)
		col1 := cond.GetArgs()[1].(*expression.Column)
		col1.ResolveIndices(joinSchema)
	}
	otherConditions := append(expression.ScalarFuncs2Exprs(v.EqualConditions), v.OtherConditions...)
	defaultValues := v.DefaultValues
	if defaultValues == nil {
		defaultValues = make([]types.Datum, v.Children()[v.SmallChildIdx].Schema().Len())
	}
	generator := newJoinResultGenerator(b.ctx, v.JoinType, v.SmallChildIdx == 0,
		defaultValues, otherConditions, leftChild.Schema().GetTypes(), rightChild.Schema().GetTypes())
	outerExec, innerExec := leftChild, rightChild
	outerFilter, innerFilter := v.LeftConditions, v.RightConditions
	if v.SmallChildIdx == 0 {
		outerExec, innerExec = rightChild, leftChild
		outerFilter, innerFilter = v.RightConditions, v.LeftConditions
	}
	e := &NestedLoopApplyExec{
		baseExecutor:    newBaseExecutor(apply.Schema(), b.ctx, outerExec, innerExec),
		innerExec:       innerExec,
		outerExec:       outerExec,
		outerFilter:     outerFilter,
		innerFilter:     innerFilter,
		outer:           v.JoinType != plan.InnerJoin,
		resultGenerator: generator,
		outerSchema:     apply.OuterSchema,
		outerChunk:      outerExec.newChunk(),
		innerChunk:      innerExec.newChunk(),
		resultChunk:     chunk.NewChunk(v.Schema().GetTypes()),
	}
	e.innerList = chunk.NewList(innerExec.Schema().GetTypes(), e.maxChunkSize)
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildExists(v *plan.PhysicalExists) Executor {
	childExec := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	e := &ExistsExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, childExec),
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildMaxOneRow(v *plan.PhysicalMaxOneRow) Executor {
	childExec := b.build(v.Children()[0])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	return &MaxOneRowExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, childExec),
	}
}

func (b *executorBuilder) buildUnionAll(v *plan.PhysicalUnionAll) Executor {
	childExecs := make([]Executor, len(v.Children()))
	for i, child := range v.Children() {
		childExecs[i] = b.build(child)
		if b.err != nil {
			b.err = errors.Trace(b.err)
			return nil
		}
	}
	e := &UnionExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, childExecs...),
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) buildUpdate(v *plan.Update) Executor {
	tblID2table := make(map[int64]table.Table)
	for id := range v.SelectPlan.Schema().TblID2Handle {
		tblID2table[id], _ = b.is.TableByID(id)
	}
	selExec := b.build(v.SelectPlan)
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	updateExec := &UpdateExec{
		baseExecutor: newBaseExecutor(nil, b.ctx, selExec),
		SelectExec:   selExec,
		OrderedList:  v.OrderedList,
		tblID2table:  tblID2table,
		IgnoreErr:    v.IgnoreErr,
	}
	updateExec.supportChk = true
	return updateExec
}

func (b *executorBuilder) buildDelete(v *plan.Delete) Executor {
	tblID2table := make(map[int64]table.Table)
	for id := range v.SelectPlan.Schema().TblID2Handle {
		tblID2table[id], _ = b.is.TableByID(id)
	}
	selExec := b.build(v.SelectPlan)
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	deleteExec := &DeleteExec{
		baseExecutor: newBaseExecutor(nil, b.ctx, selExec),
		SelectExec:   selExec,
		Tables:       v.Tables,
		IsMultiTable: v.IsMultiTable,
		tblID2Table:  tblID2table,
	}
	deleteExec.supportChk = true
	return deleteExec
}

func (b *executorBuilder) buildAnalyzeIndexPushdown(task plan.AnalyzeIndexTask) *AnalyzeIndexExec {
	e := &AnalyzeIndexExec{
		ctx:         b.ctx,
		tblInfo:     task.TableInfo,
		idxInfo:     task.IndexInfo,
		concurrency: b.ctx.GetSessionVars().IndexSerialScanConcurrency,
		priority:    b.priority,
		analyzePB: &tipb.AnalyzeReq{
			Tp:             tipb.AnalyzeType_TypeIndex,
			StartTs:        math.MaxUint64,
			Flags:          statementContextToFlags(b.ctx.GetSessionVars().StmtCtx),
			TimeZoneOffset: timeZoneOffset(b.ctx),
		},
	}
	e.analyzePB.IdxReq = &tipb.AnalyzeIndexReq{
		BucketSize: maxBucketSize,
		NumColumns: int32(len(task.IndexInfo.Columns)),
	}
	if !task.IndexInfo.Unique {
		depth := int32(defaultCMSketchDepth)
		width := int32(defaultCMSketchWidth)
		e.analyzePB.IdxReq.CmsketchDepth = &depth
		e.analyzePB.IdxReq.CmsketchWidth = &width
	}
	return e
}

func (b *executorBuilder) buildAnalyzeColumnsPushdown(task plan.AnalyzeColumnsTask) *AnalyzeColumnsExec {
	cols := task.ColsInfo
	keepOrder := false
	if task.PKInfo != nil {
		keepOrder = true
		cols = append([]*model.ColumnInfo{task.PKInfo}, cols...)
	}
	e := &AnalyzeColumnsExec{
		ctx:         b.ctx,
		tblInfo:     task.TableInfo,
		colsInfo:    task.ColsInfo,
		pkInfo:      task.PKInfo,
		concurrency: b.ctx.GetSessionVars().DistSQLScanConcurrency,
		priority:    b.priority,
		keepOrder:   keepOrder,
		analyzePB: &tipb.AnalyzeReq{
			Tp:             tipb.AnalyzeType_TypeColumn,
			StartTs:        math.MaxUint64,
			Flags:          statementContextToFlags(b.ctx.GetSessionVars().StmtCtx),
			TimeZoneOffset: timeZoneOffset(b.ctx),
		},
	}
	depth := int32(defaultCMSketchDepth)
	width := int32(defaultCMSketchWidth)
	e.analyzePB.ColReq = &tipb.AnalyzeColumnsReq{
		BucketSize:    maxBucketSize,
		SampleSize:    maxRegionSampleSize,
		SketchSize:    maxSketchSize,
		ColumnsInfo:   distsql.ColumnsToProto(cols, task.TableInfo.PKIsHandle),
		CmsketchDepth: &depth,
		CmsketchWidth: &width,
	}
	b.err = errors.Trace(setPBColumnsDefaultValue(b.ctx, e.analyzePB.ColReq.ColumnsInfo, cols))
	return e
}

func (b *executorBuilder) buildAnalyze(v *plan.Analyze) Executor {
	e := &AnalyzeExec{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
		tasks:        make([]*analyzeTask, 0, len(v.Children())),
	}
	for _, task := range v.ColTasks {
		e.tasks = append(e.tasks, &analyzeTask{
			taskType: colTask,
			colExec:  b.buildAnalyzeColumnsPushdown(task),
		})
		if b.err != nil {
			b.err = errors.Trace(b.err)
			return nil
		}
	}
	for _, task := range v.IdxTasks {
		e.tasks = append(e.tasks, &analyzeTask{
			taskType: idxTask,
			idxExec:  b.buildAnalyzeIndexPushdown(task),
		})
		if b.err != nil {
			b.err = errors.Trace(b.err)
			return nil
		}
	}
	e.supportChk = true
	return e
}

func (b *executorBuilder) constructDAGReq(plans []plan.PhysicalPlan) (*tipb.DAGRequest, error) {
	dagReq := &tipb.DAGRequest{}
	dagReq.StartTs = b.getStartTS()
	dagReq.TimeZoneOffset = timeZoneOffset(b.ctx)
	sc := b.ctx.GetSessionVars().StmtCtx
	dagReq.Flags = statementContextToFlags(sc)
	for _, p := range plans {
		execPB, err := p.ToPB(b.ctx)
		if err != nil {
			return nil, errors.Trace(err)
		}
		dagReq.Executors = append(dagReq.Executors, execPB)
	}
	return dagReq, nil
}

func (b *executorBuilder) buildIndexLookUpJoin(v *plan.PhysicalIndexJoin) Executor {
	outerExec := b.build(v.Children()[v.OuterIndex])
	if b.err != nil {
		b.err = errors.Trace(b.err)
		return nil
	}
	if outerExec.supportChunk() {
		// All inner data reader supports chunk(TableReader, IndexReader, IndexLookUpReadfer),
		// we only need to check outer.
		return b.buildNewIndexLookUpJoin(v, outerExec)
	}
	batchSize := 1
	if !v.KeepOrder {
		batchSize = b.ctx.GetSessionVars().IndexJoinBatchSize
	}
	innerExecBuilder := &dataReaderBuilder{v.Children()[1-v.OuterIndex], b}
	defaultValues := v.DefaultValues
	if defaultValues == nil {
		defaultValues = make([]types.Datum, innerExecBuilder.Schema().Len())
	}
	var outerConditions, innerConditions []expression.Expression
	if v.OuterIndex == 0 {
		outerConditions = v.LeftConditions
		innerConditions = v.RightConditions
	} else {
		outerConditions = v.RightConditions
		innerConditions = v.LeftConditions
	}
	return &IndexLookUpJoin{
		baseExecutor:     newBaseExecutor(v.Schema(), b.ctx, outerExec),
		outerExec:        outerExec,
		innerExecBuilder: innerExecBuilder,
		outerKeys:        v.OuterJoinKeys,
		innerKeys:        v.InnerJoinKeys,
		outerFilter:      outerConditions,
		innerFilter:      innerConditions,
		resultGenerator:  newJoinResultGenerator(b.ctx, v.JoinType, v.OuterIndex == 1, defaultValues, v.OtherConditions, nil, nil),
		maxBatchSize:     batchSize,
		indexRanges:      v.Ranges,
		keyOff2IdxOff:    v.KeyOff2IdxOff,
	}
}

func (b *executorBuilder) buildNewIndexLookUpJoin(v *plan.PhysicalIndexJoin, outerExec Executor) Executor {
	outerFilter, innerFilter := v.LeftConditions, v.RightConditions
	leftTypes, rightTypes := v.Children()[0].Schema().GetTypes(), v.Children()[1].Schema().GetTypes()
	outerTypes, innerTypes := leftTypes, rightTypes
	if v.OuterIndex == 1 {
		outerFilter, innerFilter = v.RightConditions, v.LeftConditions
		outerTypes, innerTypes = rightTypes, leftTypes
	}
	defaultValues := v.DefaultValues
	if defaultValues == nil {
		defaultValues = make([]types.Datum, len(innerTypes))
	}
	innerPlan := v.Children()[1-v.OuterIndex]
	e := &NewIndexLookUpJoin{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx, outerExec),
		outerCtx: outerCtx{
			rowTypes: outerTypes,
			filter:   outerFilter,
		},
		innerCtx: innerCtx{
			readerBuilder: &dataReaderBuilder{innerPlan, b},
			rowTypes:      innerTypes,
			filter:        innerFilter,
		},
		workerWg:        new(sync.WaitGroup),
		resultGenerator: newJoinResultGenerator(b.ctx, v.JoinType, v.OuterIndex == 1, defaultValues, v.OtherConditions, leftTypes, rightTypes),
		indexRanges:     v.Ranges,
		keyOff2IdxOff:   v.KeyOff2IdxOff,
	}
	e.supportChk = true
	outerKeyCols := make([]int, len(v.OuterJoinKeys))
	for i := 0; i < len(v.OuterJoinKeys); i++ {
		outerKeyCols[i] = v.OuterJoinKeys[i].Index
	}
	e.outerCtx.keyCols = outerKeyCols
	innerKeyCols := make([]int, len(v.InnerJoinKeys))
	for i := 0; i < len(v.InnerJoinKeys); i++ {
		innerKeyCols[i] = v.InnerJoinKeys[i].Index
	}
	e.innerCtx.keyCols = innerKeyCols
	e.joinResult = e.newChunk()
	return e
}

// containsLimit tests if the execs contains Limit because we do not know whether `Limit` has consumed all of its' source,
// so the feedback may not be accurate.
func containsLimit(execs []*tipb.Executor) bool {
	for _, exec := range execs {
		if exec.Limit != nil {
			return true
		}
	}
	return false
}

func buildNoRangeTableReader(b *executorBuilder, v *plan.PhysicalTableReader) (*TableReaderExecutor, error) {
	dagReq, err := b.constructDAGReq(v.TablePlans)
	if err != nil {
		return nil, errors.Trace(err)
	}
	ts := v.TablePlans[0].(*plan.PhysicalTableScan)
	table, _ := b.is.TableByID(ts.Table.ID)
	pkID := int64(-1)
	if ts.Table.PKIsHandle {
		if pk := ts.Table.GetPkColInfo(); pk != nil {
			pkID = pk.ID
		}
	}
	e := &TableReaderExecutor{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
		dagPB:        dagReq,
		tableID:      ts.Table.ID,
		table:        table,
		keepOrder:    ts.KeepOrder,
		desc:         ts.Desc,
		columns:      ts.Columns,
		priority:     b.priority,
	}
	if containsLimit(dagReq.Executors) {
		e.feedback = statistics.NewQueryFeedback(0, 0, false, 0, 0)
	} else {
		e.feedback = statistics.NewQueryFeedback(ts.Table.ID, pkID, false, ts.HistVersion, ts.StatsInfo().Count())
	}
	e.baseExecutor.supportChk = true

	for i := range v.Schema().Columns {
		dagReq.OutputOffsets = append(dagReq.OutputOffsets, uint32(i))
	}

	return e, nil
}

func (b *executorBuilder) buildTableReader(v *plan.PhysicalTableReader) *TableReaderExecutor {
	ret, err := buildNoRangeTableReader(b, v)
	if err != nil {
		b.err = errors.Trace(err)
		return nil
	}

	ts := v.TablePlans[0].(*plan.PhysicalTableScan)
	ret.ranges = ts.Ranges
	return ret
}

func buildNoRangeIndexReader(b *executorBuilder, v *plan.PhysicalIndexReader) (*IndexReaderExecutor, error) {
	dagReq, err := b.constructDAGReq(v.IndexPlans)
	if err != nil {
		return nil, errors.Trace(err)
	}
	is := v.IndexPlans[0].(*plan.PhysicalIndexScan)
	table, _ := b.is.TableByID(is.Table.ID)
	e := &IndexReaderExecutor{
		baseExecutor: newBaseExecutor(v.Schema(), b.ctx),
		dagPB:        dagReq,
		tableID:      is.Table.ID,
		table:        table,
		index:        is.Index,
		keepOrder:    !is.OutOfOrder,
		desc:         is.Desc,
		columns:      is.Columns,
		priority:     b.priority,
	}
	if containsLimit(dagReq.Executors) {
		e.feedback = statistics.NewQueryFeedback(0, 0, false, 0, 0)
	} else {
		e.feedback = statistics.NewQueryFeedback(is.Table.ID, is.Index.ID, true, is.HistVersion, is.StatsInfo().Count())
	}
	e.supportChk = true

	for _, col := range v.OutputColumns {
		dagReq.OutputOffsets = append(dagReq.OutputOffsets, uint32(col.Index))
	}

	return e, nil
}

func (b *executorBuilder) buildIndexReader(v *plan.PhysicalIndexReader) *IndexReaderExecutor {
	ret, err := buildNoRangeIndexReader(b, v)
	if err != nil {
		b.err = errors.Trace(err)
		return nil
	}

	is := v.IndexPlans[0].(*plan.PhysicalIndexScan)
	ret.ranges = is.Ranges
	return ret
}

func buildNoRangeIndexLookUpReader(b *executorBuilder, v *plan.PhysicalIndexLookUpReader) (*IndexLookUpExecutor, error) {
	indexReq, err := b.constructDAGReq(v.IndexPlans)
	if err != nil {
		return nil, errors.Trace(err)
	}
	tableReq, err := b.constructDAGReq(v.TablePlans)
	if err != nil {
		return nil, errors.Trace(err)
	}
	is := v.IndexPlans[0].(*plan.PhysicalIndexScan)
	indexReq.OutputOffsets = []uint32{uint32(len(is.Index.Columns))}
	table, _ := b.is.TableByID(is.Table.ID)

	for i := 0; i < v.Schema().Len(); i++ {
		tableReq.OutputOffsets = append(tableReq.OutputOffsets, uint32(i))
	}

	e := &IndexLookUpExecutor{
		baseExecutor:      newBaseExecutor(v.Schema(), b.ctx),
		dagPB:             indexReq,
		tableID:           is.Table.ID,
		table:             table,
		index:             is.Index,
		keepOrder:         !is.OutOfOrder,
		desc:              is.Desc,
		tableRequest:      tableReq,
		columns:           is.Columns,
		priority:          b.priority,
		dataReaderBuilder: &dataReaderBuilder{executorBuilder: b},
	}
	if containsLimit(indexReq.Executors) {
		e.feedback = statistics.NewQueryFeedback(0, 0, false, 0, 0)
	} else {
		e.feedback = statistics.NewQueryFeedback(is.Table.ID, is.Index.ID, true, is.HistVersion, is.StatsInfo().Count())
	}
	e.supportChk = true
	if cols, ok := v.Schema().TblID2Handle[is.Table.ID]; ok {
		e.handleIdx = cols[0].Index
	}
	return e, nil
}

func (b *executorBuilder) buildIndexLookUpReader(v *plan.PhysicalIndexLookUpReader) *IndexLookUpExecutor {
	ret, err := buildNoRangeIndexLookUpReader(b, v)
	if err != nil {
		b.err = errors.Trace(err)
		return nil
	}

	is := v.IndexPlans[0].(*plan.PhysicalIndexScan)
	ret.ranges = is.Ranges
	return ret
}

// dataReaderBuilder build an executor.
// The executor can be used to read data in the ranges which are constructed by datums.
// Differences from executorBuilder:
// 1. dataReaderBuilder calculate data range from argument, rather than plan.
// 2. the result executor is already opened.
type dataReaderBuilder struct {
	plan.Plan
	*executorBuilder
}

func (builder *dataReaderBuilder) buildExecutorForIndexJoin(goCtx goctx.Context, datums [][]types.Datum,
	IndexRanges []*ranger.NewRange, keyOff2IdxOff []int) (Executor, error) {
	switch v := builder.Plan.(type) {
	case *plan.PhysicalTableReader:
		return builder.buildTableReaderForIndexJoin(goCtx, v, datums)
	case *plan.PhysicalIndexReader:
		return builder.buildIndexReaderForIndexJoin(goCtx, v, datums, IndexRanges, keyOff2IdxOff)
	case *plan.PhysicalIndexLookUpReader:
		return builder.buildIndexLookUpReaderForIndexJoin(goCtx, v, datums, IndexRanges, keyOff2IdxOff)
	}
	return nil, errors.New("Wrong plan type for dataReaderBuilder")
}

func (builder *dataReaderBuilder) buildTableReaderForIndexJoin(goCtx goctx.Context, v *plan.PhysicalTableReader, datums [][]types.Datum) (Executor, error) {
	e, err := buildNoRangeTableReader(builder.executorBuilder, v)
	if err != nil {
		return nil, errors.Trace(err)
	}
	handles := make([]int64, 0, len(datums))
	for _, datum := range datums {
		handles = append(handles, datum[0].GetInt64())
	}
	return builder.buildTableReaderFromHandles(goCtx, e, handles)
}

func (builder *dataReaderBuilder) buildTableReaderFromHandles(goCtx goctx.Context, e *TableReaderExecutor, handles []int64) (Executor, error) {
	sort.Sort(sortutil.Int64Slice(handles))
	var b requestBuilder
	kvReq, err := b.SetTableHandles(e.tableID, handles).
		SetDAGRequest(e.dagPB).
		SetDesc(e.desc).
		SetKeepOrder(e.keepOrder).
		SetPriority(e.priority).
		SetFromSessionVars(e.ctx.GetSessionVars()).
		Build()
	if err != nil {
		return nil, errors.Trace(err)
	}
	e.result, err = distsql.SelectDAG(goCtx, builder.ctx, kvReq, e.schema.GetTypes())
	if err != nil {
		return nil, errors.Trace(err)
	}
	e.result.Fetch(goCtx)
	return e, nil
}

func (builder *dataReaderBuilder) buildIndexReaderForIndexJoin(goCtx goctx.Context, v *plan.PhysicalIndexReader,
	values [][]types.Datum, indexRanges []*ranger.NewRange, keyOff2IdxOff []int) (Executor, error) {
	e, err := buildNoRangeIndexReader(builder.executorBuilder, v)
	if err != nil {
		return nil, errors.Trace(err)
	}
	kvRanges, err := buildKvRangesForIndexJoin(e.ctx.GetSessionVars().StmtCtx, e.tableID, e.index.ID, values, indexRanges, keyOff2IdxOff)
	if err != nil {
		return nil, errors.Trace(err)
	}
	err = e.open(goCtx, kvRanges)
	return e, errors.Trace(err)
}

func (builder *dataReaderBuilder) buildIndexLookUpReaderForIndexJoin(goCtx goctx.Context, v *plan.PhysicalIndexLookUpReader,
	values [][]types.Datum, indexRanges []*ranger.NewRange, keyOff2IdxOff []int) (Executor, error) {
	e, err := buildNoRangeIndexLookUpReader(builder.executorBuilder, v)
	if err != nil {
		return nil, errors.Trace(err)
	}
	kvRanges, err := buildKvRangesForIndexJoin(e.ctx.GetSessionVars().StmtCtx, e.tableID, e.index.ID, values, indexRanges, keyOff2IdxOff)
	if err != nil {
		return nil, errors.Trace(err)
	}
	err = e.open(goCtx, kvRanges)
	return e, errors.Trace(err)
}

// buildKvRangesForIndexJoin builds kv ranges for index join when the inner plan is index scan plan.
func buildKvRangesForIndexJoin(sc *stmtctx.StatementContext, tableID, indexID int64, keyDatums [][]types.Datum, indexRanges []*ranger.NewRange, keyOff2IdxOff []int) ([]kv.KeyRange, error) {
	kvRanges := make([]kv.KeyRange, 0, len(indexRanges)*len(keyDatums))
	for _, val := range keyDatums {
		for _, ran := range indexRanges {
			for keyOff, idxOff := range keyOff2IdxOff {
				ran.LowVal[idxOff] = val[keyOff]
				ran.HighVal[idxOff] = val[keyOff]
			}
		}
		tmpKvRanges, err := indexRangesToKVRanges(sc, tableID, indexID, indexRanges)
		if err != nil {
			return nil, errors.Trace(err)
		}
		kvRanges = append(kvRanges, tmpKvRanges...)
	}
	// kvRanges don't overlap each other. So compare StartKey is enough.
	sort.Slice(kvRanges, func(i, j int) bool {
		return bytes.Compare(kvRanges[i].StartKey, kvRanges[j].StartKey) < 0
	})
	return kvRanges, nil
}


text:
<?php

namespace CO\EShopBundle\Entity;

use Doctrine\ORM\EntityRepository;

/**
 * OrderRepository
 *
 * This class was generated by the Doctrine ORM. Add your own custom
 * repository methods below.
 */
class OrderRepository extends EntityRepository
{

    public function getItems($status)
    {
		$crit = array();
		if ($status<>0) {
			$crit = array('status'=>$status);
		}

        return $this->findBy($crit, array('id' => 'DESC'));
    }

}

redpajama_set_name: RedPajamaGithub


text:
from rdopkg.exception import CommandFailed
from rdopkg.exception import RdopkgException

def test_exception_default():
    e = RdopkgException()
    assert str(e) == 'An unknown error occurred'
    assert e.exit_code == 1

def test_exception_msg_provided():
    e = RdopkgException(msg='foo bar')
    assert str(e) == 'foo bar'

def test_CommandFailed_bad_kwargs():
    e = CommandFailed(foo='bar')
    assert str(e) == e.msg_fmt

def test_CommandFailed_good_kwargs():
    e = CommandFailed(cmd='bar')
    assert str(e) == e.msg_fmt % {'cmd': 'bar'}

def test_exception_exit_code():
    for code in [-128, -1, 0, 1, 23, 1337]:
        e = RdopkgException(exit_code=code)
        assert e.exit_code == code


text:
using System.Threading.Tasks;
using Blogifier.Core.Data.Models;
using System.Net.Http;

namespace Blogifier.Core.Services.Syndication.Rss
{
    public interface IRssService
    {
        Task<HttpResponseMessage> Import(RssImportModel model);
        string Display(string absoluteUri, string author);
    }
}


text:
package com.codepath.apps.twittertweets;

import android.content.Context;

/*
 * This is the Android application itself and is used to configure various settings
 * including the image cache in memory and on disk. This also adds a singleton
 * for accessing the relevant rest client.
 *
 *     TwitterClient client = TwitterApplication.getRestClient();
 *     // use client to send requests to API
 *
 */
public class TwitterApplication extends com.activeandroid.app.Application {
	private static Context context;

	@Override
	public void onCreate() {
		super.onCreate();
		TwitterApplication.context = this;
	}

	public static TwitterClient getRestClient() {
		return (TwitterClient) TwitterClient.getInstance(TwitterClient.class, TwitterApplication.context);
	}
}


text:
import { xxxxx } from './utils';

// // eslint-disable-next-line
console.log(1);
console.log(2);
console.log(xxxxx);


text:
import json

import requests
from six import string_types

#probably to be wrapped by some stateful client object

#request session
s = None
def _session():
    global s
    if s is None:
        s = requests.session()
    return s

def _reset_session():
    global s
    s = None

def sanitize_url(url):
    if not url.endswith("/"):
        url += "/"
    return url

class MBSClientException(Exception):
    pass

def _request(method, url, data=None, files=None, headers=None, session=None):
    if session is None:
        session = _session()
    resp = session.request(method, url, data=data, files=files, headers=headers)
    if resp.status_code != 200:
        raise MBSClientException(resp.status_code)
    else:
        return resp.json()

def register(root_url, username, password, session=None):
    raise NotImplementedError

def login(root_url, username, password, session=None):
    raise NotImplementedError

def upload(root_url, file_or_path, session=None):
    url = sanitize_url(root_url) + "upload"
    if isinstance(file_or_path, string_types):
        with open(file_or_path, 'rb') as f:
            _request('POST', url, files={'file' : f})
    else:
        _request('POST', url, files={'file' : file_or_path})

def ls(root_url, username=None):
    root_url = sanitize_url(root_url)
    if username:
        url =  root_url + "ls/%s" % username
    else:
        url = root_url + "ls"
    return _request('GET', url)

def configure(root_url, uri, _delete=False, **kwargs):
    url = sanitize_url(root_url) + "configure"
    data = json.dumps(dict(uri=uri, kwargs=kwargs, _delete=_delete))
    return _request('POST', url, data=data,
                    headers={'content-type' : 'application/json'})


text:
<?xml version="1.0" encoding="utf-8"?>
<manifest xmlns:android="http://schemas.android.com/apk/res/android"
    package="cn.huaxingtan.player"
    android:versionCode="2013111100"
    android:installLocation="preferExternal"
    android:versionName="1.0" >

    <uses-sdk
        android:minSdkVersion="14"
        android:targetSdkVersion="18" />

    <uses-permission android:name="android.permission.INTERNET" >
    </uses-permission>
    <uses-permission android:name="android.permission.WAKE_LOCK" >
    </uses-permission>
    <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE"></uses-permission>
	<uses-permission android:name="android.permission.READ_PHONE_STATE"></uses-permission>

	<uses-permission android:name="android.permission.ACCESS_WIFI_STATE" />
	<uses-permission android:name="android.permission.WRITE_EXTERNAL_STORAGE" />

    <application
        android:allowBackup="true"
        android:icon="@drawable/ic_152"
        android:label="@string/app_name"
        android:theme="@style/AppTheme" >
        <activity
            android:name="cn.huaxingtan.view.MainActivity"
            android:configChanges="keyboardHidden|orientation"
            android:label="@string/app_name" >
            <intent-filter>
                <action android:name="android.intent.action.MAIN" />
                <category android:name="android.intent.category.LAUNCHER" />
            </intent-filter>
        </activity>
        <activity
            android:name="cn.huaxingtan.view.MusicPlayerActivity"
            android:screenOrientation="portrait"
            android:label="@string/title_activity_music_player">
        </activity>
        <activity
            android:name="cn.huaxingtan.view.DetailActivity"
            android:configChanges="keyboardHidden|orientation"
            android:label="@string/app_name" >
        </activity>
        <activity
            android:name="cn.huaxingtan.view.DownloadingActivity"
            android:configChanges="keyboardHidden|orientation"
            android:label="@string/download_actvity_name" >
        </activity>
        <activity android:name="com.umeng.fb.ConversationActivity"/>
		<activity android:name="com.umeng.fb.ContactActivity"/>
        <service
            android:name="cn.huaxingtan.service.MusicPlayerService">
        </service>
        <service
		    android:name="com.umeng.common.net.DownloadingService"
		    android:process=":DownloadingService" >
		</service>
        <receiver android:name="cn.huaxingtan.receiver.DownloadManagerReceiver">
            <intent-filter>
                <action android:name="android.intent.action.DOWNLOAD_COMPLETE"/>
            </intent-filter>
        </receiver>
        <meta-data android:value="527faca556240bf08b0888c0" android:name="UMENG_APPKEY"></meta-data>
        <meta-data android:value="1" android:name="UMENG_CHANNEL"/>
    </application>

</manifest>


text:
@interface HETChooseDeviceViewController : HETBaseViewController

@end


text:
var gha = require('github'),
    twit = require('twitter'),
    fs = require('fs'),
    util = require('util'),
    expand = require('expand-home-dir'),
    request = require('request'),
    stats = require('text-statistics'),
    twitter_config_path = expand('~/.twitter.api.json'),
    no_post_message = '@RussellHay should be ashamed. He hasn\'t written in %d days. http://github.com/RussTheAerialist/writing',
    statistics_message = '@RussellHay wrote stuff today which has a reading grade level of %s (SMOG): %s',
    github_user = process.env.npm_package_config_github_user || 'RussTheAerialist',
    github_repo = process.env.npm_package_config_repo || 'writing'
    ;

var gh = new gha({
       version: '3.0.0'
    });

var now = new Date(process.argv[2] || undefined);

function post_twitter_message_real(message) {
    fs.readFile(twitter_config_path, function(err, data) {
        if (err) { throw err; }

        var client = new twit(JSON.parse(data));
        client.post('statuses/update', {
            status: message
        }, function(err, tweet, response) {
            if (err) { throw err; }
            console.log(tweet, response);
        });
    });
}

var post_twitter_message = post_twitter_message_real;
function post_twitter_message_debug(message) {
	console.log(message);
}

if (process.env.npm_package_config_debug === '1') {
	post_twitter_message = post_twitter_message_debug;
}

function get_last_commit_url(callback) {
	gh.repos.getCommits({
		user: github_user,
		repo: github_repo,
		page: 0,
		per_page: 1
	}, function (err, res) {
		if (err) {
			callback(err, res);
			return;
		}

		var commit = res[0],
		    sha = commit.sha,
		    html_url = commit.html_url;

		gh.repos.getCommit({
			user: github_user,
			repo: github_repo,
			sha: sha
		}, function (err, res) {
			if (err) {
				callback(err, res);
				return;
			}

			var file_url = res.files[0].raw_url;
			callback(undefined, file_url, html_url);
		});
	});
}

function do_statistics_post() {
	get_last_commit_url(function (err, url, commit_url) {
		if (err) { throw err; }

		request(url, function (err, response, body) {
			if (err) { throw err; }
			if (response.statusCode != 200) {
				throw response;
			}

			var calc = new stats(body);
			var message = util.format(statistics_message, calc.smogIndex(), commit_url);
			post_twitter_message(message);
		});
	});
}

gh.repos.get({
    user: github_user,
    repo: github_repo
}, function(err, res) {
    var last_push = new Date(res.pushed_at);
    var days_since_last_push = Math.floor((now - last_push)/(86400000)); // (1000 * 3600 * 24)
    if (days_since_last_push > 1) {
        post_twitter_message(util.format(no_post_message, days_since_last_push));
    } else {
    	do_statistics_post();
    }
});


text:
package com.amazonaws.services.mediaconnect.model;

import java.io.Serializable;
import javax.annotation.Generated;
import com.amazonaws.protocol.StructuredPojo;
import com.amazonaws.protocol.ProtocolMarshaller;

/**
 * Attributes related to the transport stream that are used in a source or output.
 *
 * @see AWS API
 *      Documentation
 */
@Generated("com.amazonaws:aws-java-sdk-code-generator")
public class Transport implements Serializable, Cloneable, StructuredPojo {

    /**
     * The range of IP addresses that should be allowed to initiate output requests to this flow. These IP addresses
     * should be in the form of a Classless Inter-Domain Routing (CIDR) block; for example, 10.0.0.0/16.
     */
    private java.util.List<String> cidrAllowList;
    /** The smoothing max bitrate for RIST, RTP, and RTP-FEC streams. */
    private Integer maxBitrate;
    /**
     * The maximum latency in milliseconds. This parameter applies only to RIST-based, Zixi-based, and Fujitsu-based
     * streams.
     */
    private Integer maxLatency;
    /** The size of the buffer (in milliseconds) to use to sync incoming source data. */
    private Integer maxSyncBuffer;
    /**
     * The minimum latency in milliseconds for SRT-based streams. In streams that use the SRT protocol, this value that
     * you set on your MediaConnect source or output represents the minimal potential latency of that connection. The
     * latency of the stream is set to the highest number between the sender's minimum latency and the receiver's minimum
     * latency.
     */
    private Integer minLatency;
    /** The protocol that is used by the source or output. */
    private String protocol;
    /** The remote ID for the Zixi-pull stream. */
    private String remoteId;
    /** The port that the flow uses to send outbound requests to initiate connection with the sender. */
    private Integer senderControlPort;
    /** The IP address that the flow communicates with to initiate connection with the sender. */
    private String senderIpAddress;
    /** The smoothing latency in milliseconds for RIST, RTP, and RTP-FEC streams. */
    private Integer smoothingLatency;
    /** Source IP or domain name for SRT-caller protocol. */
    private String sourceListenerAddress;
    /** Source port for SRT-caller protocol. */
    private Integer sourceListenerPort;
    /** The stream ID that you want to use for this transport. This parameter applies only to Zixi-based streams. */
    private String streamId;

    /**
     * The range of IP addresses that should be allowed to initiate output requests to this flow. These IP addresses
     * should be in the form of a Classless Inter-Domain Routing (CIDR) block; for example, 10.0.0.0/16.
     *
     * @return The range of IP addresses that should be allowed to initiate output requests to this flow. These IP
     *         addresses should be in the form of a Classless Inter-Domain Routing (CIDR) block; for example,
     *         10.0.0.0/16.
     */

    public java.util.List<String> getCidrAllowList() {
        return cidrAllowList;
    }

    /**
     * The range of IP addresses that should be allowed to initiate output requests to this flow. These IP addresses
     * should be in the form of a Classless Inter-Domain Routing (CIDR) block; for example, 10.0.0.0/16.
     *
     * @param cidrAllowList
     *        The range of IP addresses that should be allowed to initiate output requests to this flow. These IP
     *        addresses should be in the form of a Classless Inter-Domain Routing (CIDR) block; for example,
     *        10.0.0.0/16.
     */

    public void setCidrAllowList(java.util.Collection<String> cidrAllowList) {
        if (cidrAllowList == null) {
            this.cidrAllowList = null;
            return;
        }

        this.cidrAllowList = new java.util.ArrayList<String>(cidrAllowList);
    }

    /**
     * The range of IP addresses that should be allowed to initiate output requests to this flow. These IP addresses
     * should be in the form of a Classless Inter-Domain Routing (CIDR) block; for example, 10.0.0.0/16.
     *

     * NOTE: This method appends the values to the existing list (if any). Use
     * {@link #setCidrAllowList(java.util.Collection)} or {@link #withCidrAllowList(java.util.Collection)} if you want
     * to override the existing values.
     *

     *
     * @param cidrAllowList
     *        The range of IP addresses that should be allowed to initiate output requests to this flow. These IP
     *        addresses should be in the form of a Classless Inter-Domain Routing (CIDR) block; for example,
     *        10.0.0.0/16.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withCidrAllowList(String... cidrAllowList) {
        if (this.cidrAllowList == null) {
            setCidrAllowList(new java.util.ArrayList<String>(cidrAllowList.length));
        }
        for (String ele : cidrAllowList) {
            this.cidrAllowList.add(ele);
        }
        return this;
    }

    /**
     * The range of IP addresses that should be allowed to initiate output requests to this flow. These IP addresses
     * should be in the form of a Classless Inter-Domain Routing (CIDR) block; for example, 10.0.0.0/16.
     *
     * @param cidrAllowList
     *        The range of IP addresses that should be allowed to initiate output requests to this flow. These IP
     *        addresses should be in the form of a Classless Inter-Domain Routing (CIDR) block; for example,
     *        10.0.0.0/16.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withCidrAllowList(java.util.Collection<String> cidrAllowList) {
        setCidrAllowList(cidrAllowList);
        return this;
    }

    /**
     * The smoothing max bitrate for RIST, RTP, and RTP-FEC streams.
     *
     * @param maxBitrate
     *        The smoothing max bitrate for RIST, RTP, and RTP-FEC streams.
     */

    public void setMaxBitrate(Integer maxBitrate) {
        this.maxBitrate = maxBitrate;
    }

    /**
     * The smoothing max bitrate for RIST, RTP, and RTP-FEC streams.
     *
     * @return The smoothing max bitrate for RIST, RTP, and RTP-FEC streams.
     */

    public Integer getMaxBitrate() {
        return this.maxBitrate;
    }

    /**
     * The smoothing max bitrate for RIST, RTP, and RTP-FEC streams.
     *
     * @param maxBitrate
     *        The smoothing max bitrate for RIST, RTP, and RTP-FEC streams.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withMaxBitrate(Integer maxBitrate) {
        setMaxBitrate(maxBitrate);
        return this;
    }

    /**
     * The maximum latency in milliseconds. This parameter applies only to RIST-based, Zixi-based, and Fujitsu-based
     * streams.
     *
     * @param maxLatency
     *        The maximum latency in milliseconds. This parameter applies only to RIST-based, Zixi-based, and
     *        Fujitsu-based streams.
     */

    public void setMaxLatency(Integer maxLatency) {
        this.maxLatency = maxLatency;
    }

    /**
     * The maximum latency in milliseconds. This parameter applies only to RIST-based, Zixi-based, and Fujitsu-based
     * streams.
     *
     * @return The maximum latency in milliseconds. This parameter applies only to RIST-based, Zixi-based, and
     *         Fujitsu-based streams.
     */

    public Integer getMaxLatency() {
        return this.maxLatency;
    }

    /**
     * The maximum latency in milliseconds. This parameter applies only to RIST-based, Zixi-based, and Fujitsu-based
     * streams.
     *
     * @param maxLatency
     *        The maximum latency in milliseconds. This parameter applies only to RIST-based, Zixi-based, and
     *        Fujitsu-based streams.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withMaxLatency(Integer maxLatency) {
        setMaxLatency(maxLatency);
        return this;
    }

    /**
     * The size of the buffer (in milliseconds) to use to sync incoming source data.
     *
     * @param maxSyncBuffer
     *        The size of the buffer (in milliseconds) to use to sync incoming source data.
     */

    public void setMaxSyncBuffer(Integer maxSyncBuffer) {
        this.maxSyncBuffer = maxSyncBuffer;
    }

    /**
     * The size of the buffer (in milliseconds) to use to sync incoming source data.
     *
     * @return The size of the buffer (in milliseconds) to use to sync incoming source data.
     */

    public Integer getMaxSyncBuffer() {
        return this.maxSyncBuffer;
    }

    /**
     * The size of the buffer (in milliseconds) to use to sync incoming source data.
     *
     * @param maxSyncBuffer
     *        The size of the buffer (in milliseconds) to use to sync incoming source data.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withMaxSyncBuffer(Integer maxSyncBuffer) {
        setMaxSyncBuffer(maxSyncBuffer);
        return this;
    }

    /**
     * The minimum latency in milliseconds for SRT-based streams. In streams that use the SRT protocol, this value that
     * you set on your MediaConnect source or output represents the minimal potential latency of that connection. The
     * latency of the stream is set to the highest number between the sender's minimum latency and the receiver's minimum
     * latency.
     *
     * @param minLatency
     *        The minimum latency in milliseconds for SRT-based streams. In streams that use the SRT protocol, this
     *        value that you set on your MediaConnect source or output represents the minimal potential latency of that
     *        connection. The latency of the stream is set to the highest number between the sender's minimum latency
     *        and the receiver's minimum latency.
     */

    public void setMinLatency(Integer minLatency) {
        this.minLatency = minLatency;
    }

    /**
     * The minimum latency in milliseconds for SRT-based streams. In streams that use the SRT protocol, this value that
     * you set on your MediaConnect source or output represents the minimal potential latency of that connection. The
     * latency of the stream is set to the highest number between the sender's minimum latency and the receiver's minimum
     * latency.
     *
     * @return The minimum latency in milliseconds for SRT-based streams. In streams that use the SRT protocol, this
     *         value that you set on your MediaConnect source or output represents the minimal potential latency of that
     *         connection. The latency of the stream is set to the highest number between the sender's minimum latency
     *         and the receiver's minimum latency.
     */

    public Integer getMinLatency() {
        return this.minLatency;
    }

    /**
     * The minimum latency in milliseconds for SRT-based streams. In streams that use the SRT protocol, this value that
     * you set on your MediaConnect source or output represents the minimal potential latency of that connection. The
     * latency of the stream is set to the highest number between the sender's minimum latency and the receiver's minimum
     * latency.
     *
     * @param minLatency
     *        The minimum latency in milliseconds for SRT-based streams. In streams that use the SRT protocol, this
     *        value that you set on your MediaConnect source or output represents the minimal potential latency of that
     *        connection. The latency of the stream is set to the highest number between the sender's minimum latency
     *        and the receiver's minimum latency.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withMinLatency(Integer minLatency) {
        setMinLatency(minLatency);
        return this;
    }

    /**
     * The protocol that is used by the source or output.
     *
     * @param protocol
     *        The protocol that is used by the source or output.
     * @see Protocol
     */

    public void setProtocol(String protocol) {
        this.protocol = protocol;
    }

    /**
     * The protocol that is used by the source or output.
     *
     * @return The protocol that is used by the source or output.
     * @see Protocol
     */

    public String getProtocol() {
        return this.protocol;
    }

    /**
     * The protocol that is used by the source or output.
     *
     * @param protocol
     *        The protocol that is used by the source or output.
     * @return Returns a reference to this object so that method calls can be chained together.
     * @see Protocol
     */

    public Transport withProtocol(String protocol) {
        setProtocol(protocol);
        return this;
    }

    /**
     * The protocol that is used by the source or output.
     *
     * @param protocol
     *        The protocol that is used by the source or output.
     * @return Returns a reference to this object so that method calls can be chained together.
     * @see Protocol
     */

    public Transport withProtocol(Protocol protocol) {
        this.protocol = protocol.toString();
        return this;
    }

    /**
     * The remote ID for the Zixi-pull stream.
     *
     * @param remoteId
     *        The remote ID for the Zixi-pull stream.
     */

    public void setRemoteId(String remoteId) {
        this.remoteId = remoteId;
    }

    /**
     * The remote ID for the Zixi-pull stream.
     *
     * @return The remote ID for the Zixi-pull stream.
     */

    public String getRemoteId() {
        return this.remoteId;
    }

    /**
     * The remote ID for the Zixi-pull stream.
     *
     * @param remoteId
     *        The remote ID for the Zixi-pull stream.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withRemoteId(String remoteId) {
        setRemoteId(remoteId);
        return this;
    }

    /**
     * The port that the flow uses to send outbound requests to initiate connection with the sender.
     *
     * @param senderControlPort
     *        The port that the flow uses to send outbound requests to initiate connection with the sender.
     */

    public void setSenderControlPort(Integer senderControlPort) {
        this.senderControlPort = senderControlPort;
    }

    /**
     * The port that the flow uses to send outbound requests to initiate connection with the sender.
     *
     * @return The port that the flow uses to send outbound requests to initiate connection with the sender.
     */

    public Integer getSenderControlPort() {
        return this.senderControlPort;
    }

    /**
     * The port that the flow uses to send outbound requests to initiate connection with the sender.
     *
     * @param senderControlPort
     *        The port that the flow uses to send outbound requests to initiate connection with the sender.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withSenderControlPort(Integer senderControlPort) {
        setSenderControlPort(senderControlPort);
        return this;
    }

    /**
     * The IP address that the flow communicates with to initiate connection with the sender.
     *
     * @param senderIpAddress
     *        The IP address that the flow communicates with to initiate connection with the sender.
     */

    public void setSenderIpAddress(String senderIpAddress) {
        this.senderIpAddress = senderIpAddress;
    }

    /**
     * The IP address that the flow communicates with to initiate connection with the sender.
     *
     * @return The IP address that the flow communicates with to initiate connection with the sender.
     */

    public String getSenderIpAddress() {
        return this.senderIpAddress;
    }

    /**
     * The IP address that the flow communicates with to initiate connection with the sender.
     *
     * @param senderIpAddress
     *        The IP address that the flow communicates with to initiate connection with the sender.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withSenderIpAddress(String senderIpAddress) {
        setSenderIpAddress(senderIpAddress);
        return this;
    }

    /**
     * The smoothing latency in milliseconds for RIST, RTP, and RTP-FEC streams.
     *
     * @param smoothingLatency
     *        The smoothing latency in milliseconds for RIST, RTP, and RTP-FEC streams.
     */

    public void setSmoothingLatency(Integer smoothingLatency) {
        this.smoothingLatency = smoothingLatency;
    }

    /**
     * The smoothing latency in milliseconds for RIST, RTP, and RTP-FEC streams.
     *
     * @return The smoothing latency in milliseconds for RIST, RTP, and RTP-FEC streams.
     */

    public Integer getSmoothingLatency() {
        return this.smoothingLatency;
    }

    /**
     * The smoothing latency in milliseconds for RIST, RTP, and RTP-FEC streams.
     *
     * @param smoothingLatency
     *        The smoothing latency in milliseconds for RIST, RTP, and RTP-FEC streams.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withSmoothingLatency(Integer smoothingLatency) {
        setSmoothingLatency(smoothingLatency);
        return this;
    }

    /**
     * Source IP or domain name for SRT-caller protocol.
     *
     * @param sourceListenerAddress
     *        Source IP or domain name for SRT-caller protocol.
     */

    public void setSourceListenerAddress(String sourceListenerAddress) {
        this.sourceListenerAddress = sourceListenerAddress;
    }

    /**
     * Source IP or domain name for SRT-caller protocol.
     *
     * @return Source IP or domain name for SRT-caller protocol.
     */

    public String getSourceListenerAddress() {
        return this.sourceListenerAddress;
    }

    /**
     * Source IP or domain name for SRT-caller protocol.
     *
     * @param sourceListenerAddress
     *        Source IP or domain name for SRT-caller protocol.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withSourceListenerAddress(String sourceListenerAddress) {
        setSourceListenerAddress(sourceListenerAddress);
        return this;
    }

    /**
     * Source port for SRT-caller protocol.
     *
     * @param sourceListenerPort
     *        Source port for SRT-caller protocol.
     */

    public void setSourceListenerPort(Integer sourceListenerPort) {
        this.sourceListenerPort = sourceListenerPort;
    }

    /**
     * Source port for SRT-caller protocol.
     *
     * @return Source port for SRT-caller protocol.
     */

    public Integer getSourceListenerPort() {
        return this.sourceListenerPort;
    }

    /**
     * Source port for SRT-caller protocol.
     *
     * @param sourceListenerPort
     *        Source port for SRT-caller protocol.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withSourceListenerPort(Integer sourceListenerPort) {
        setSourceListenerPort(sourceListenerPort);
        return this;
    }

    /**
     * The stream ID that you want to use for this transport. This parameter applies only to Zixi-based streams.
     *
     * @param streamId
     *        The stream ID that you want to use for this transport. This parameter applies only to Zixi-based streams.
     */

    public void setStreamId(String streamId) {
        this.streamId = streamId;
    }

    /**
     * The stream ID that you want to use for this transport. This parameter applies only to Zixi-based streams.
     *
     * @return The stream ID that you want to use for this transport. This parameter applies only to Zixi-based streams.
     */

    public String getStreamId() {
        return this.streamId;
    }

    /**
     * The stream ID that you want to use for this transport. This parameter applies only to Zixi-based streams.
     *
     * @param streamId
     *        The stream ID that you want to use for this transport. This parameter applies only to Zixi-based streams.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public Transport withStreamId(String streamId) {
        setStreamId(streamId);
        return this;
    }

    /**
     * Returns a string representation of this object. This is useful for testing and debugging. Sensitive data will be
     * redacted from this string using a placeholder value.
     *
     * @return A string representation of this object.
     *
     * @see java.lang.Object#toString()
     */
    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("{");
        if (getCidrAllowList() != null)
            sb.append("CidrAllowList: ").append(getCidrAllowList()).append(",");
        if (getMaxBitrate() != null)
            sb.append("MaxBitrate: ").append(getMaxBitrate()).append(",");
        if (getMaxLatency() != null)
            sb.append("MaxLatency: ").append(getMaxLatency()).append(",");
        if (getMaxSyncBuffer() != null)
            sb.append("MaxSyncBuffer: ").append(getMaxSyncBuffer()).append(",");
        if (getMinLatency() != null)
            sb.append("MinLatency: ").append(getMinLatency()).append(",");
        if (getProtocol() != null)
            sb.append("Protocol: ").append(getProtocol()).append(",");
        if (getRemoteId() != null)
            sb.append("RemoteId: ").append(getRemoteId()).append(",");
        if (getSenderControlPort() != null)
            sb.append("SenderControlPort: ").append(getSenderControlPort()).append(",");
        if (getSenderIpAddress() != null)
            sb.append("SenderIpAddress: ").append(getSenderIpAddress()).append(",");
        if (getSmoothingLatency() != null)
            sb.append("SmoothingLatency: ").append(getSmoothingLatency()).append(",");
        if (getSourceListenerAddress() != null)
            sb.append("SourceListenerAddress: ").append(getSourceListenerAddress()).append(",");
        if (getSourceListenerPort() != null)
            sb.append("SourceListenerPort: ").append(getSourceListenerPort()).append(",");
        if (getStreamId() != null)
            sb.append("StreamId: ").append(getStreamId());
        sb.append("}");
        return sb.toString();
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;

        if (obj instanceof Transport == false)
            return false;
        Transport other = (Transport) obj;
        if (other.getCidrAllowList() == null ^ this.getCidrAllowList() == null)
            return false;
        if (other.getCidrAllowList() != null && other.getCidrAllowList().equals(this.getCidrAllowList()) == false)
            return false;
        if (other.getMaxBitrate() == null ^ this.getMaxBitrate() == null)
            return false;
        if (other.getMaxBitrate() != null && other.getMaxBitrate().equals(this.getMaxBitrate()) == false)
            return false;
        if (other.getMaxLatency() == null ^ this.getMaxLatency() == null)
            return false;
        if (other.getMaxLatency() != null && other.getMaxLatency().equals(this.getMaxLatency()) == false)
            return false;
        if (other.getMaxSyncBuffer() == null ^ this.getMaxSyncBuffer() == null)
            return false;
        if (other.getMaxSyncBuffer() != null && other.getMaxSyncBuffer().equals(this.getMaxSyncBuffer()) == false)
            return false;
        if (other.getMinLatency() == null ^ this.getMinLatency() == null)
            return false;
        if (other.getMinLatency() != null && other.getMinLatency().equals(this.getMinLatency()) == false)
            return false;
        if (other.getProtocol() == null ^ this.getProtocol() == null)
            return false;
        if (other.getProtocol() != null && other.getProtocol().equals(this.getProtocol()) == false)
            return false;
        if (other.getRemoteId() == null ^ this.getRemoteId() == null)
            return false;
        if (other.getRemoteId() != null && other.getRemoteId().equals(this.getRemoteId()) == false)
            return false;
        if (other.getSenderControlPort() == null ^ this.getSenderControlPort() == null)
            return false;
        if (other.getSenderControlPort() != null && other.getSenderControlPort().equals(this.getSenderControlPort()) == false)
            return false;
        if (other.getSenderIpAddress() == null ^ this.getSenderIpAddress() == null)
            return false;
        if (other.getSenderIpAddress() != null && other.getSenderIpAddress().equals(this.getSenderIpAddress()) == false)
            return false;
        if (other.getSmoothingLatency() == null ^ this.getSmoothingLatency() == null)
            return false;
        if (other.getSmoothingLatency() != null && other.getSmoothingLatency().equals(this.getSmoothingLatency()) == false)
            return false;
        if (other.getSourceListenerAddress() == null ^ this.getSourceListenerAddress() == null)
            return false;
        if (other.getSourceListenerAddress() != null && other.getSourceListenerAddress().equals(this.getSourceListenerAddress()) == false)
            return false;
        if (other.getSourceListenerPort() == null ^ this.getSourceListenerPort() == null)
            return false;
        if (other.getSourceListenerPort() != null && other.getSourceListenerPort().equals(this.getSourceListenerPort()) == false)
            return false;
        if (other.getStreamId() == null ^ this.getStreamId() == null)
            return false;
        if (other.getStreamId() != null && other.getStreamId().equals(this.getStreamId()) == false)
            return false;
        return true;
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int hashCode = 1;

        hashCode = prime * hashCode + ((getCidrAllowList() == null) ? 0 : getCidrAllowList().hashCode());
        hashCode = prime * hashCode + ((getMaxBitrate() == null) ? 0 : getMaxBitrate().hashCode());
        hashCode = prime * hashCode + ((getMaxLatency() == null) ? 0 : getMaxLatency().hashCode());
        hashCode = prime * hashCode + ((getMaxSyncBuffer() == null) ? 0 : getMaxSyncBuffer().hashCode());
        hashCode = prime * hashCode + ((getMinLatency() == null) ? 0 : getMinLatency().hashCode());
        hashCode = prime * hashCode + ((getProtocol() == null) ? 0 : getProtocol().hashCode());
        hashCode = prime * hashCode + ((getRemoteId() == null) ? 0 : getRemoteId().hashCode());
        hashCode = prime * hashCode + ((getSenderControlPort() == null) ? 0 : getSenderControlPort().hashCode());
        hashCode = prime * hashCode + ((getSenderIpAddress() == null) ? 0 : getSenderIpAddress().hashCode());
        hashCode = prime * hashCode + ((getSmoothingLatency() == null) ? 0 : getSmoothingLatency().hashCode());
        hashCode = prime * hashCode + ((getSourceListenerAddress() == null) ? 0 : getSourceListenerAddress().hashCode());
        hashCode = prime * hashCode + ((getSourceListenerPort() == null) ? 0 : getSourceListenerPort().hashCode());
        hashCode = prime * hashCode + ((getStreamId() == null) ? 0 : getStreamId().hashCode());
        return hashCode;
    }

    @Override
    public Transport clone() {
        try {
            return (Transport) super.clone();
        } catch (CloneNotSupportedException e) {
            throw new IllegalStateException("Got a CloneNotSupportedException from Object.clone() " + "even though we're Cloneable!", e);
        }
    }

    @com.amazonaws.annotation.SdkInternalApi
    @Override
    public void marshall(ProtocolMarshaller protocolMarshaller) {
        com.amazonaws.services.mediaconnect.model.transform.TransportMarshaller.getInstance().marshall(this, protocolMarshaller);
    }
}


text:
"""
https://github.com/behave/behave/issues/767

When trying to do something like::

    fixture_registry = {'fixture.foo': foo_fixture}
    f = use_fixture_by_tag('fixture.foo', context, fixture_registry)

Behave returns nothing. ::

    repr(f)
    'None'

This seems to be an oversight.
"""

from mock import Mock
from behave.fixture import fixture, use_fixture_by_tag
from behave.runner import Context

def test_issue_767_use_feature_by_tag_has_no_return():
    """Verifies that issue #767 is fixed."""

    @fixture(name='fixture.foo')
    def foo_fixture(context, *args, **kwargs):
        context.foo = 'foo'
        return context.foo

    # -- SCHEMA 1: fixture_func
    fixture_registry1 = {
        "fixture.foo": foo_fixture
    }
    # -- SCHEMA 2: fixture_func, fixture_args, fixture_kwargs
    fixture_registry2 = {
        "fixture.foo": (foo_fixture, (), {})
    }

    context = Context(runner=Mock())
    fixture1 = use_fixture_by_tag("fixture.foo", context, fixture_registry1)
    assert fixture1 == "foo"
    assert context.foo is fixture1

    context = Context(runner=Mock())
    fixture2 = use_fixture_by_tag("fixture.foo", context, fixture_registry2)
    assert fixture2 == "foo"
    assert context.foo is fixture2


text:
"""prtg_wan_2: module called query PRTG API for bandwidth statistics

NOTES:
* PRTG API is GET only and doesn't use sessions, so we pass the username and passhash on each connection.
* We're using the table.json function to retrieve multiple sensor values at once. We filter the results by
using a tag we've assigned to the sensors in PRTG called "statusboard_wan."

* The JSON output from this module is still formatted in the old iPad Statusboard app format:

{"graph":
    {"title": "WAN Bandwidth (mbps)", "type": "line", "refreshEveryNSeconds": 60,
    "datasequences": [
    {"title": "RAL",
        "datapoints": [{"title": "17:11", "value": 0.0}, {"title": "17:17", "value": 0.0}, ....
    {"title": "TRI",
        "datapoints": [{"title": "17:11", "value": 0.0}, {"title": "17:12", "value": 0.0}, ....

"""
import time
import json
import logging.config
import requests
from credentials import PRTG_USERNAME
from credentials import PRTG_PASSHASH
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

__author__ = 'scott@flakshack.com (Scott Vintinner)'

# =================================SETTINGS======================================
MAX_DATAPOINTS = 120
SAMPLE_INTERVAL = 15
GRAPH_TITLE = "WAN (mbps)"
PRTG_AUTH = "&username=" + PRTG_USERNAME + "&passhash=" + PRTG_PASSHASH
PRTG_URL = ("https://prtg/api/table.json?" +
            "content=sensors&columns=objid,lastvalue&filter_tags=@tag(statusboard_wan2)")

# #### PRTG SENSORS #####
# The sensors are retrieved by a tag filter, but we'll use this list to help name them.
# id:  This is the PRTG sensor ID (displayed on the sensor web page or URL)
# name:  This is the name of the device as it will appear on the graph
PRTG_SENSORS = (
    {"objid": 7407, "name": "PRI-SL2"},
    {"objid": 7406, "name": "PRI-L3"},
    {"objid": 7796, "name": "DR-SL2"},
    {"objid": 7493, "name": "DR-L3"},
    {"objid": 7405, "name": "PRI-W"}
)

# ================================================================================

class MonitorJSON:
    """This is a simple class passed to Monitor threads so we can access the current JSON data in that thread"""
    def __init__(self):
        self.json = output_message("Loading...", "")

class PRTGSensor:
    """We create a single object of this class and store all our data in it."""
    all_sensors = []    # Static array containing all sensors

    def __init__(self, sensor_id, sensor_name):
        self.objid = sensor_id
        self.name = sensor_name
        self.datapoints = []                 # Hold raw values from PRTG
        self.__class__.all_sensors.append(self)     # Add self to static array

class PRTGPausedException(Exception):
    def __init__(self, value):
        self.value = value

    def __str__(self):
        return repr(self.value)

class PRTGRequestException(Exception):
    pass

def output_message(message, detail):
    """This function will output an error message formatted in JSON to display on the StatusBoard app"""
    statusbar_output = {"graph": {"title": GRAPH_TITLE, "error": {"message": message, "detail": detail}}}
    output = json.dumps(statusbar_output)
    return output

def generate_json(prtg_monitor):
    """This function will connect to the PRTG server API and store the output in prtg_monitor.json"""

    logger = logging.getLogger("prtg_interface_2")
    time_x_axis = time.strftime("%H:%M")  # The time that we'll add to the X axis of the chart

    # Create a list of PRTGSensors using the contants provided above (we'll store the data in this object)
    if len(PRTGSensor.all_sensors) == 0:    # This is only done once
        for sensor in PRTG_SENSORS:
            PRTGSensor(sensor["objid"], sensor["name"])

    try:
        # ############ PRTG API CALL ###############
        logger.debug("Getting: " + PRTG_URL)
        r = requests.get(PRTG_URL + PRTG_AUTH, verify=False)
        if r.status_code != 200:
            raise PRTGRequestException("Request error status: " + str(r.status_code) + " " + r.reason)
        api_data = r.json()  # Convert returned byte stream to json

        # Loop through our list of sensors, match the object id with the item returned by the API call
        # and add a datapoint to our array.
        for sensor in PRTGSensor.all_sensors:
            for api_sensor in api_data["sensors"]:
                if api_sensor['lastvalue'] == '-':
                    raise PRTGPausedException(sensor.name)
                if sensor.objid == api_sensor["objid"]:
                    # The lastvalue_raw field is the bps divided by 8.  It doesn't make any sense
                    # why they store it this way instead of just storing the bits.  Seriously, WTF?
                    # The conversion from raw to mbps is lastvalue_raw*8/(1000*1000)
                    # Note we're using SI (decimal) notation here because that is what PRTG uses.
                    mbps = round(((api_sensor["lastvalue_raw"])*8)/1000000, 2)  # Convert from PRTG raw to mbps
                    sensor.datapoints.append({"title": time_x_axis, "value": mbps})

                    # If we already have the max number of datapoints, delete the oldest item.
                    if len(sensor.datapoints) >= MAX_DATAPOINTS:
                        del(sensor.datapoints[0])

        # #### Format the JSON data that is expected by the javascript front-end #####
        statusbar_datasequences = []
        for sensor in PRTGSensor.all_sensors:
            statusbar_datasequences.append({"title": sensor.name, "datapoints": sensor.datapoints})

        statusbar_graph = {
            "title": GRAPH_TITLE, "type": "line",
            "refreshEveryNSeconds": SAMPLE_INTERVAL,
            "datasequences": statusbar_datasequences
        }
        statusbar_type = {"graph": statusbar_graph}
        prtg_monitor.json = json.dumps(statusbar_type)

    except PRTGPausedException as sensor_name:

        logger.error("PRTG sensor " + str(sensor_name) + " is paused")
        prtg_monitor.json = output_message("PRTG sensor " + str(sensor_name) + " is paused", '')
        PRTGSensor.all_sensors = []  # Reset the saved data

    except Exception as error:
        logger.error("Error getting data from PRTG: " + str(error))
        prtg_monitor.json = output_message("Error getting data from PRTG", str(error))
        PRTGSensor.all_sensors = []  # Reset the saved data

    logger.debug(prtg_monitor.json)

# ======================================================
# __main__
#
# If you run this module by itself, it will instantiate
# the MonitorJSON class and start an infinite loop
# printing data.
# ======================================================
#
if __name__ == '__main__':

    # When run by itself, we need to create the logger object (which is normally created in webserver.py)
    try:
        f = open("log_settings.json", 'rt')
        log_config = json.load(f)
        f.close()
        logging.config.dictConfig(log_config)
    except FileNotFoundError as e:
        print("Log configuration file not found: " + str(e))
        logging.basicConfig(level=logging.DEBUG)        # fallback to basic settings
    except json.decoder.JSONDecodeError as e:
        print("Error parsing logger config file: " + str(e))
        raise

    monitor = MonitorJSON()
    while True:
        main_logger = logging.getLogger(__name__)
        generate_json(monitor)
        # Wait X seconds for the next iteration
        main_logger.debug("Waiting for " + str(SAMPLE_INTERVAL) + " seconds")
        time.sleep(SAMPLE_INTERVAL)


text:
"""
Copyright (c) 2004-Present Pivotal Software, Inc.

This program and the accompanying materials are made available under
the terms of the 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.
"""

import os
import shlex
import sys
import ConfigParser
import StringIO
import unittest2 as unittest

from discovery import TINCDiscoveryQueryHandler

try:
    from unittest2.signals import installHandler
except ImportError:
    installHandler = None

from tinctest.loader import TINCTestLoader
from tinctest.case import TINCTestCase

class TINCException(Exception):
    """
    Base class for all tinc exceptions.
    """
    pass

class TINCSchedule(object):
    """
    Encapsulates the parameters given in a schedule file
    Schedule files are searched in the following folders:
      - if not, fully qualified name search for the file in
          - TINCREPOHOME/schedules
          - PWD
    """
    def __init__(self, schedule_file):
        self.schedule_file = schedule_file
        self.schedule_name = None
        self.time_limit = None
        self.options = None
        self._parse_schedule_file()

    def _parse_schedule_file(self):
        if not self.schedule_file:
            raise TINCException('schedule file name not given')

        file_to_parse = None
        if os.path.isabs(self.schedule_file) and os.path.exists(self.schedule_file):
            file_to_parse = self.schedule_file
        else:
            if os.environ.has_key("TINC_SCHEDULE_DIR"):
                file_env_folder = os.path.join(os.environ["TINC_SCHEDULE_DIR"], self.schedule_file)
                if os.path.exists(file_env_folder):
                    file_to_parse = file_env_folder

            if not file_to_parse:
                if os.environ.has_key("PWD"):
                    file_sched_folder = os.path.join(os.environ["PWD"], 'schedules', self.schedule_file)
                    if os.path.exists(file_sched_folder):
                        file_to_parse = file_sched_folder
                    if not file_to_parse:
                        file_curdir = os.path.join(os.environ["PWD"], self.schedule_file)
                        if os.path.exists(file_curdir):
                            file_to_parse = file_curdir

        if not file_to_parse:
            raise TINCException("schedule file [%s] doesn't exist" %self.schedule_file)

        cfg_string = '[Schedule]\n'
        cfg_string = cfg_string + open(file_to_parse).read()

        fd = StringIO.StringIO(cfg_string)
        cfg = ConfigParser.RawConfigParser()
        cfg.readfp(fd)

        for key, val in cfg.items('Schedule'):
            low_key = key.lower()
            if low_key == 'name':
                self.schedule_name = val
            elif low_key == 'timelimit':
                self.time_limit = int(val)
            elif low_key == 'options':
                self.options = os.path.expandvars(val)
            else:
                #logger not configured;
                raise TINCException("unknown key [%s] in schedule file [%s]" %(key, self.schedule_file))

class TINCTestProgram(unittest.TestProgram):

    tinc_config = None

    def parseArgs(self, argv):
        """
        TINCTestProgram accepts some additional arguments that should be propogated to tests
        before they are run. This parses the additional arguments after calling out to
        the super class.
        """
        if len(argv) > 1 and argv[1].lower() == 'discover':
            self._do_discovery(argv[2:])
            return

        import getopt
        long_opts = ['help', 'verbose', 'quiet', 'failfast', 'catch', 'buffer', 'schedule']
        # Add custom options
        long_opts[len(long_opts):] = ['codeline=','testconfig=']
        try:
            options, args = getopt.getopt(argv[1:], 'hHvqfcbs:', long_opts)
            for opt, value in options:
                if opt in ('-h','-H','--help'):
                    self.usageExit()
                if opt in ('-q','--quiet'):
                    self.verbosity = 0
                if opt in ('-v','--verbose'):
                    self.verbosity = 2
                if opt in ('-f','--failfast'):
                    if self.failfast is None:
                        self.failfast = True
                    # Should this raise an exception if -f is not valid?
                if opt in ('-c','--catch'):
                    if self.catchbreak is None and installHandler is not None:
                        self.catchbreak = True
                    # Should this raise an exception if -c is not valid?
                if opt in ('-b','--buffer'):
                    if self.buffer is None:
                        self.buffer = True
                    # Should this raise an exception if -b is not valid?
                if opt in ('-s','--schedule'):
                    self.schedule = value
                    schedule_obj = TINCSchedule(self.schedule)
                    if not schedule_obj:
                        raise TINCExcpetion("Schedule file %s doesn't list any options to tinc.py" %self.schedule)
                    self.parseArgs([argv[0]] + shlex.split(schedule_obj.options))
                    return

                if opt in ('--testconfig'):
                    TINCTestCase.test_config.extend(value.split(','))

            if len(args) == 0 and self.defaultTest is None:
                # createTests will load tests from self.module
                self.testNames = None
            elif len(args) > 0:
                self.testNames = args
                if __name__ == '__main__':
                    # to support python -m unittest ...
                    self.module = None
            else:
                self.testNames = (self.defaultTest,)
            self.createTests()
        except getopt.error, msg:
            self.usageExit(msg)

    def _do_discovery(self, argv, Loader=TINCTestLoader):
        """
        TINCTestProgram provides a minimialistic change to _do_discovery, in which
        the Loader is made to be a TINCTestLoader. Without this change, TestProgram
        would use the base TestLoader, incapable of discovering and loading SQLTestCases.

        This code is copy/paste-d from unittest.TestProgram, simply because deferring
        to the parent in this instance was found to be cumbersome.
        """
        import optparse
        # handle command line args for test discovery
        self.progName = '%s discover' % self.progName

        # Following options are related to the running of tests through discovery
        options_list = []
        options_list.append(optparse.make_option('-v', '--verbose', dest='verbose', default=False,
                                                 help='Verbose output', action='store_true'))

        if self.failfast != False:
            options_list.append(optparse.make_option('-f', '--failfast', dest='failfast', default=False,
                                                     help='Stop on first fail or error',
                                                     action='store_true'))
        if self.catchbreak != False and installHandler is not None:
            options_list.append(optparse.make_option('-c', '--catch', dest='catchbreak', default=False,
                                                     help='Catch ctrl-C and display results so far',
                                                     action='store_true'))
        if self.buffer != False:
            options_list.append(optparse.make_option('-b', '--buffer', dest='buffer', default=False,
                                                     help='Buffer stdout and stderr during tests',
                                                     action='store_true'))

        test_suite, options = TINCTestProgram.construct_discover_test_suite(argv, Loader=TINCTestLoader,
                                                                            additional_discover_options = options_list,
                                                                            progName = self.progName
                                                                            )

        # only set options from the parsing here
        # if they weren't set explicitly in the constructor
        if self.failfast is None:
            self.failfast = options.failfast
        if self.catchbreak is None and installHandler is not None:
            self.catchbreak = options.catchbreak
        if self.buffer is None:
            self.buffer = options.buffer

        if options.verbose:
            self.verbosity = 2
        self.test = test_suite

    @staticmethod
    def construct_discover_test_suite(argv,Loader=TINCTestLoader, additional_discover_options = [], progName=''):
        import optparse
        parser = optparse.OptionParser()
        parser.prog = progName

        parser.add_option('-s', '--start-directory', dest='start', default=[], action='append',
                          help="Directory to start discovery ('.' default), can be specified multiple times to run discovery from different folders.")
        parser.add_option('-t', '--top-level-directory', dest='top', default=None,
                          help='Top level directory of project (defaults to start directory)')
        parser.add_option('-p', '--pattern', dest='pattern', default=[], action='append',
                          help="Module patterns to discover tests from('test*.py' default), can be specified multiple " + \
                          "times to discover tests from multiple patterns.")

        parser.add_option('-q', '--query', dest='queries', action='append', default=None,
                          help='TINC test queries to filter discovered tests. Can be specified multiple times to provide multiple queries. Refer to ' + \
                          'documentation for more information on the queries supported.')

        parser.add_option('-d', '--dryrun', dest='dryrun', default=False,
                          help='No tests will be executed. Only test discovery will happen.',
                          action='store_true')

        for option in additional_discover_options:
            parser.add_option(option)

        options, args = parser.parse_args(argv)

        if len(args) > 2 :
            parser.print_help()
            sys.exit(1)

        for name, value in zip(('start', 'pattern'), args):
            setattr(options, name, [value])

        if not options.start:
            options.start = ['.']
        if not options.pattern:
            options.pattern = ['test*']

        start_dirs = options.start
        patterns = [x.strip('\'"') for x in options.pattern]
        top_level_dir = options.top

        loader = Loader()

        if options.queries:
            tinc_queries = [x.strip('\'"') for x in options.queries]
            query_handler = TINCDiscoveryQueryHandler(tinc_queries)
        else:
            query_handler = None
        test_suite = loader.discover(start_dirs, patterns, top_level_dir, query_handler, options.dryrun)
        return test_suite, options


text:
<!DOCTYPE html>

<title>geolocation test</title>
<link rel="stylesheet" type="text/css" href="style.css"/>

NOT TESTED (script did not run)

Test setup: Set opera:config#Geolocation|LocationProviderURL to