// buildExecutor build a executor from plan, prepared statement may need additional procedure.
420708
func (a *ExecStmt) buildExecutor(ctx sessionctx.Context) (Executor, error) {
	if _, ok := a.Plan.(*plannercore.Execute); !ok {
		// Do not sync transaction for Execute statement, because the real optimization work is done in
		// "ExecuteExec.Build".
		useMaxTS, err := IsPointGetWithPKOrUniqueKeyByAutoCommit(ctx, a.Plan)
		if err != nil {
			return nil, err
		}
		if useMaxTS {
			logutil.Logger(context.Background()).Debug("init txnStartTS with MaxUint64", zap.Uint64("conn", ctx.GetSessionVars().ConnectionID), zap.String("text", a.Text))
			err = ctx.InitTxnWithStartTS(math.MaxUint64)
		} else if ctx.GetSessionVars().SnapshotTS != 0 {
			if _, ok := a.Plan.(*plannercore.CheckTable); ok {
				err = ctx.InitTxnWithStartTS(ctx.GetSessionVars().SnapshotTS)
			}
		}
		if err != nil {
			return nil, err
		}

		stmtCtx := ctx.GetSessionVars().StmtCtx
		if stmtPri := stmtCtx.Priority; stmtPri == mysql.NoPriority {
			switch {
			case useMaxTS:
				stmtCtx.Priority = kv.PriorityHigh
			case a.Expensive:
				stmtCtx.Priority = kv.PriorityLow
			}
		}
	}
	if _, ok := a.Plan.(*plannercore.Analyze); ok && ctx.GetSessionVars().InRestrictedSQL {
		ctx.GetSessionVars().StmtCtx.Priority = kv.PriorityLow
	}

	b := newExecutorBuilder(ctx, a.InfoSchema)
	e := b.build(a.Plan)
	if b.err != nil {
		return nil, errors.Trace(b.err)
	}

	// ExecuteExec is not a real Executor, we only use it to build another Executor from a prepared statement.
	if executorExec, ok := e.(*ExecuteExec); ok {
		err := executorExec.Build()
		if err != nil {
			return nil, err
		}
		a.isPreparedStmt = true
		a.Plan = executorExec.plan
		e = executorExec.stmtExec
	}
	return e, nil
}
1218662
func NewExecutor(ctx *plan.Context, planner plan.Planner) *JobExecutor {
	e := &JobExecutor{}
	e.Executor = e
	e.Planner = planner
	e.Ctx = ctx
	return e
}
361622
func CreateExecutor(logger hclog.Logger, driverConfig *base.ClientDriverConfig,
	executorConfig *ExecutorConfig) (Executor, *plugin.Client, error) {

	c, err := json.Marshal(executorConfig)
	if err != nil {
		return nil, nil, fmt.Errorf("unable to create executor config: %v", err)
	}
	bin, err := discover.NomadExecutable()
	if err != nil {
		return nil, nil, fmt.Errorf("unable to find the nomad binary: %v", err)
	}

	p := &ExecutorPlugin{
		logger:      logger,
		fsIsolation: executorConfig.FSIsolation,
	}

	config := &plugin.ClientConfig{
		HandshakeConfig:  base.Handshake,
		Plugins:          map[string]plugin.Plugin{"executor": p},
		Cmd:              exec.Command(bin, "executor", string(c)),
		AllowedProtocols: []plugin.Protocol{plugin.ProtocolGRPC},
		Logger:           logger.Named("executor"),
	}

	if driverConfig != nil {
		config.MaxPort = driverConfig.ClientMaxPort
		config.MinPort = driverConfig.ClientMinPort
	} else {
		config.MaxPort = ExecutorDefaultMaxPort
		config.MinPort = ExecutorDefaultMinPort
	}

	// setting the setsid of the plugin process so that it doesn't get signals sent to
	// the nomad client.
	if config.Cmd != nil {
		isolateCommand(config.Cmd)
	}

	return newExecutorClient(config, logger)
}
1087602
func (b *executorBuilder) buildMergeJoin(v *plannercore.PhysicalMergeJoin) Executor {
	leftExec := b.build(v.Children()[0])
	if b.err != nil {
		return nil
	}

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

	defaultValues := v.DefaultValues
	if defaultValues == nil {
		if v.JoinType == plannercore.RightOuterJoin {
			defaultValues = make([]types.Datum, leftExec.Schema().Len())
		} else {
			defaultValues = make([]types.Datum, rightExec.Schema().Len())
		}
	}

	e := &MergeJoinExec{
		stmtCtx:      b.ctx.GetSessionVars().StmtCtx,
		baseExecutor: newBaseExecutor(b.ctx, v.Schema(), v.ExplainID(), leftExec, rightExec),
		compareFuncs: v.CompareFuncs,
		joiner: newJoiner(
			b.ctx,
			v.JoinType,
			v.JoinType == plannercore.RightOuterJoin,
			defaultValues,
			v.OtherConditions,
			leftExec.retTypes(),
			rightExec.retTypes(),
		),
		isOuterJoin: v.JoinType.IsOuterJoin(),
	}

	leftKeys := v.LeftKeys
	rightKeys := v.RightKeys

	e.outerIdx = 0
	innerFilter := v.RightConditions

	e.innerTable = &mergeJoinInnerTable{
		reader:   rightExec,
		joinKeys: rightKeys,
	}

	e.outerTable = &mergeJoinOuterTable{
		reader: leftExec,
		filter: v.LeftConditions,
		keys:   leftKeys,
	}

	if v.JoinType == plannercore.RightOuterJoin {
		e.outerIdx = 1
		e.outerTable.reader = rightExec
		e.outerTable.filter = v.RightConditions
		e.outerTable.keys = rightKeys

		innerFilter = v.LeftConditions
		e.innerTable.reader = leftExec
		e.innerTable.joinKeys = leftKeys
	}

	// optimizer should guarantee that filters on inner table are pushed down
	// to tikv or extracted to a Selection.
	if len(innerFilter) != 0 {
		b.err = errors.Annotate(ErrBuildExecutor, "merge join's inner filter should be empty.")
		return nil
	}

	executorCounterMergeJoinExec.Inc()
	return e
}
1101649
func planBQL(ctx context.Context, bql string, s storage.Store, chanSize, bulkSize int, w io.Writer) (planner.Executor, error) {
	bql = strings.TrimSpace(bql)
	if bql == ";" {
		tracer.Trace(w, func() []string {
			return []string{"Empty statement found"}
		})
		return nil, nil
	}
	p, err := grammar.NewParser(grammar.SemanticBQL())
	if err != nil {
		msg := fmt.Errorf("NewParser failed; %v", err)
		tracer.Trace(w, func() []string {
			return []string{msg.Error()}
		})
		return nil, msg
	}
	stm := &semantic.Statement{}
	if err := p.Parse(grammar.NewLLk(bql, 1), stm); err != nil {
		msg := fmt.Errorf("NewLLk parser failed; %v", err)
		tracer.Trace(w, func() []string {
			return []string{msg.Error()}
		})
		return nil, msg
	}
	pln, err := planner.New(ctx, s, stm, chanSize, bulkSize, w)
	if err != nil {
		msg := fmt.Errorf("planer.New failed failed; %v", err)
		tracer.Trace(w, func() []string {
			return []string{msg.Error()}
		})
		return nil, msg
	}
	tracer.Trace(w, func() []string {
		return []string{"Plan successfuly created"}
	})
	return pln, nil
}
1319568
func build(expr string) (q query, err error) {
	defer func() {
		if e := recover(); e != nil {
			switch x := e.(type) {
			case string:
				err = errors.New(x)
			case error:
				err = x
			default:
				err = errors.New("unknown panic")
			}
		}
	}()
	root := parse(expr)
	b := &builder{}
	return b.processNode(root)
}
989683
func (s *HTTPServer) preparedQueryExplain(id string, resp http.ResponseWriter, req *http.Request) (interface{}, error) {
	args := structs.PreparedQueryExecuteRequest{
		QueryIDOrName: id,
		Agent: structs.QuerySource{
			Node:       s.agent.config.NodeName,
			Datacenter: s.agent.config.Datacenter,
			Segment:    s.agent.config.SegmentName,
		},
	}
	s.parseSource(req, &args.Source)
	if done := s.parse(resp, req, &args.Datacenter, &args.QueryOptions); done {
		return nil, nil
	}
	if err := parseLimit(req, &args.Limit); err != nil {
		return nil, fmt.Errorf("Bad limit: %s", err)
	}

	var reply structs.PreparedQueryExplainResponse
	defer setMeta(resp, &reply.QueryMeta)
RETRY_ONCE:
	if err := s.agent.RPC("PreparedQuery.Explain", &args, &reply); err != nil {
		// We have to check the string since the RPC sheds
		// the specific error type.
		if err.Error() == consul.ErrQueryNotFound.Error() {
			resp.WriteHeader(http.StatusNotFound)
			fmt.Fprint(resp, err.Error())
			return nil, nil
		}
		return nil, err
	}
	if args.QueryOptions.AllowStale && args.MaxStaleDuration > 0 && args.MaxStaleDuration < reply.LastContact {
		args.AllowStale = false
		args.MaxStaleDuration = 0
		goto RETRY_ONCE
	}
	reply.ConsistencyLevel = args.QueryOptions.ConsistencyLevel()
	return reply, nil
}
145343
func NewExecutor(logger hclog.Logger) Executor {
	logger = logger.Named("executor")
	if err := shelpers.Init(); err != nil {
		logger.Error("unable to initialize stats", "error", err)
	}
	return &UniversalExecutor{
		logger:         logger,
		processExited:  make(chan interface{}),
		totalCpuStats:  stats.NewCpuStats(),
		userCpuStats:   stats.NewCpuStats(),
		systemCpuStats: stats.NewCpuStats(),
		pidCollector:   newPidCollector(logger),
	}
}
121633
func (m *JobExecutor) WalkPreparedStatement(p *plan.PreparedStatement) (Task, error) {
	return nil, ErrNotImplemented
}
101342
func (e *InExp) Build(db *DB, params Params) string {
	if len(e.values) == 0 {
		if e.not {
			return ""
		}
		return "0=1"
	}

	var values []string
	for _, value := range e.values {
		switch value.(type) {
		case nil:
			values = append(values, "NULL")
		case Expression:
			sql := value.(Expression).Build(db, params)
			values = append(values, sql)
		default:
			name := fmt.Sprintf("p%v", len(params))
			params[name] = value
			values = append(values, "{:"+name+"}")
		}
	}
	col := db.QuoteColumnName(e.col)
	if len(values) == 1 {
		if e.not {
			return col + "<>" + values[0]
		}
		return col + "=" + values[0]
	}
	in := "IN"
	if e.not {
		in = "NOT IN"
	}
	return fmt.Sprintf("%v %v (%v)", col, in, strings.Join(values, ", "))
}
1143320
func NewPlanBuilder(sctx sessionctx.Context, is infoschema.InfoSchema) *PlanBuilder {
	return &PlanBuilder{
		ctx:       sctx,
		is:        is,
		colMapper: make(map[*ast.ColumnNameExpr]int),
	}
}
1076741
func (db *DB) MustExecute(stmt Executable, args ...interface{}) sql.Result {
	s, params, err := db.compile(stmt)
	if err != nil {
		log.Panicf(
			"aspect: failed to compile (%s): %s",
			stmt,
			err,
		)
	}
	if len(args) == 0 {
		args = params.args
	}
	result, err := db.conn.Exec(s, args...)
	if err != nil {
		log.Panicf(
			"aspect: failed to exec (%s) with parameters (%v): %s",
			s,
			args,
			err,
		)
	}
	return result
}
456538
func (builder *TsQueryCommandBuilder) Build() (Command, error) {
	if builder.protobuf == nil {
		panic("builder.protobuf must not be nil")
	}

	if len(builder.protobuf.GetQuery().GetBase()) == 0 {
		return nil, ErrQueryRequired
	}

	if builder.protobuf.GetStream() && builder.callback == nil {
		return nil, newClientError("TsQueryCommand requires a callback when streaming.", nil)
	}

	return &TsQueryCommand{
		protobuf: builder.protobuf,
		callback: builder.callback,
	}, nil
}
1081580
func (p *ProductBuilder) Build() {
	elapsed := rand.Intn(p.bound)

	time.Sleep(time.Duration(elapsed) * time.Millisecond)

	p.totalTime.MustInc(float64(elapsed))
	p.completed.Up()
}
183414
func New(ctx context.Context, store storage.Store, stm *semantic.Statement, chanSize, bulkSize int, w io.Writer) (Executor, error) {
	switch stm.Type() {
	case semantic.Query:
		return newQueryPlan(ctx, store, stm, chanSize, w)
	case semantic.Insert:
		return &insertPlan{
			stm:    stm,
			store:  store,
			tracer: w,
		}, nil
	case semantic.Delete:
		return &deletePlan{
			stm:    stm,
			store:  store,
			tracer: w,
		}, nil
	case semantic.Create:
		return &createPlan{
			stm:    stm,
			store:  store,
			tracer: w,
		}, nil
	case semantic.Drop:
		return &dropPlan{
			stm:    stm,
			store:  store,
			tracer: w,
		}, nil
	case semantic.Construct:
		qp, _ := newQueryPlan(ctx, store, stm, chanSize, w)
		return &constructPlan{
			stm:       stm,
			store:     store,
			tracer:    w,
			bulkSize:  bulkSize,
			queryPlan: qp,
			construct: true,
		}, nil
	case semantic.Deconstruct:
		qp, _ := newQueryPlan(ctx, store, stm, chanSize, w)
		return &constructPlan{
			stm:       stm,
			store:     store,
			tracer:    w,
			bulkSize:  bulkSize,
			queryPlan: qp,
			construct: false,
		}, nil
	case semantic.Show:
		return &showPlan{
			stm:    stm,
			store:  store,
			tracer: w,
		}, nil
	default:
		return nil, fmt.Errorf("planner.New: unknown statement type in statement %v", stm)
	}
}
1366659
func (b *Builder) Build(p *Params) {
	go b.registerSignalHandler()
	go func() {
		// used for triggering the first build
		b.watcher.update <- struct{}{}
	}()

	for range b.watcher.Wait() {
		fileName := p.generateBinaryName()

		pkg := p.packagePath()

		log.Println("build started")
		color.Cyan("Building %s...\n", pkg)

		// build package
		cmd, err := runCommand("go", "build", "-i", "-o", fileName, pkg)
		if err != nil {
			log.Fatalf("Could not run 'go build' command: %s", err)
			continue
		}

		if err := cmd.Wait(); err != nil {
			if err := interpretError(err); err != nil {
				color.Red("An error occurred while building: %s", err)
			} else {
				color.Red("A build error occurred. Please update your code...")
			}

			continue
		}
		log.Println("build completed")

		// and start the new process
		b.runner.restart(fileName)
	}
}
416538
func (b *PlanBuilder) buildExplainFor(explainFor *ast.ExplainForStmt) (Plan, error) {
	processInfo, ok := b.ctx.GetSessionManager().GetProcessInfo(explainFor.ConnectionID)
	if !ok {
		return nil, ErrNoSuchThread.GenWithStackByArgs(explainFor.ConnectionID)
	}
	if b.ctx.GetSessionVars() != nil && b.ctx.GetSessionVars().User != nil {
		if b.ctx.GetSessionVars().User.Username != processInfo.User {
			err := ErrAccessDenied.GenWithStackByArgs(b.ctx.GetSessionVars().User.Username, b.ctx.GetSessionVars().User.Hostname)
			// Different from MySQL's behavior and document.
			b.visitInfo = appendVisitInfo(b.visitInfo, mysql.SuperPriv, "", "", "", err)
		}
	}

	targetPlan, ok := processInfo.Plan.(Plan)
	if !ok || targetPlan == nil {
		return &Explain{Format: explainFor.Format}, nil
	}

	return b.buildExplainPlan(targetPlan, explainFor.Format, false, nil)
}