// TaskExecute enqueues a task for execution.
978714
func (s *globalTaskService) TaskEnqueue(
	ctx types.Context,
	run types.TaskRunFunc,
	schema []byte) *types.Task {

	t := newGenericTask(ctx, run, schema)
	go func() { execTask(t) }()
	return &t.Task
}
126384
func (vf *VindexFunc) StreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
	r, err := vf.mapVindex(vcursor, bindVars)
	if err != nil {
		return err
	}
	if err := callback(&sqltypes.Result{Fields: r.Fields}); err != nil {
		return err
	}
	return callback(&sqltypes.Result{Rows: r.Rows})
}
569859
func (m *JobExecutor) NewTask(p plan.Task) Task {
	if p.IsParallel() {
		return NewTaskParallel(m.Ctx)
	}
	return NewTaskSequential(m.Ctx)
}
223172
func (s *globalTaskService) TaskInspect(taskID int) *types.Task {
	s.RLock()
	defer s.RUnlock()
	if t, ok := s.tasks[taskID]; ok {
		return &t.Task
	}
	return nil
}
785651
func (s *Scheduler) ExecuteTaskGroup(ctx context.Context,
	fc *flow.Flow,
	taskGroupStatus *pb.FlowExecutionStatus_TaskGroup,
	wg *sync.WaitGroup,
	taskGroup *plan.TaskGroup,
	bid float64, relatedFiles []resource.FileResource) {

	defer wg.Done()

	tasks := taskGroup.Tasks
	lastTask := tasks[len(tasks)-1]
	if tasks[0].Step.IsOnDriverSide {
		// these should be only one task on the driver side
		if err := taskGroupStatus.Track(func(exeStatus *pb.FlowExecutionStatus_TaskGroup_Execution) error {
			return s.localExecute(ctx, fc, exeStatus, lastTask, wg)
		}); err != nil {
			log.Fatalf("Failed to execute on driver side: %v", err)
		}
		return
	}
	if !needsInputFromDriver(tasks[0]) {
		// wait until inputs are registed
		s.shardLocator.waitForInputDatasetShardLocations(tasks[0])
	}
	if isInputOnDisk(tasks[0]) && !isRestartableTasks(tasks) {
		// for non-restartable taskGroup, wait until on disk inputs are completed
		for _, stepGroup := range taskGroup.ParentStepGroup.Parents {
			stepGroup.WaitForAllTasksToComplete()
		}
	}

	// fmt.Printf("inputs of %s is %s\n", tasks[0].Name(), s.allInputLocations(tasks[0]))

	pickedServerChan := make(chan market.Supply, 1)
	s.Market.AddDemand(market.Requirement(taskGroup), bid, pickedServerChan)

	// get assigned executor location
	supply := <-pickedServerChan
	allocation := supply.Object.(*pb.Allocation)
	defer s.Market.ReturnSupply(supply)

	if needsInputFromDriver(tasks[0]) {
		// tell the driver to write to me
		for _, shard := range tasks[0].InputShards {
			// println("registering", shard.Name(), "at", allocation.Location.URL())
			s.setShardLocation(shard, pb.DataLocation{
				Name:     shard.Name(),
				Location: allocation.Location,
				OnDisk:   shard.Dataset.GetIsOnDiskIO(),
			})
		}
	}

	for _, shard := range lastTask.OutputShards {
		// println("registering", shard.Name(), "at", allocation.Location.URL(), "onDisk", shard.Dataset.GetIsOnDiskIO())
		s.setShardLocation(shard, pb.DataLocation{
			Name:     shard.Name(),
			Location: allocation.Location,
			OnDisk:   shard.Dataset.GetIsOnDiskIO(),
		})
	}

	// send driver code only when using go mapper reducer
	var hasGoCode bool
	for _, t := range tasks {
		hasGoCode = hasGoCode || t.Step.IsGoCode
	}
	if hasGoCode {
		relatedFiles = append(relatedFiles, resource.FileResource{os.Args[0], "."})
	}

	if len(relatedFiles) > 0 {
		err := withClient(allocation.Location.URL(), func(client pb.GleamAgentClient) error {
			for _, relatedFile := range relatedFiles {
				err := sendRelatedFile(ctx, client, fc.HashCode, relatedFile)
				if err != nil {
					taskGroup.MarkStop(err)
					return err
				}
			}
			return nil
		})
		if err != nil {
			log.Fatalf("Failed to send related files: %v", err)
		}
	}

	fn := func() error {
		err := taskGroupStatus.Track(func(exeStatus *pb.FlowExecutionStatus_TaskGroup_Execution) error {
			return s.remoteExecuteOnLocation(ctx, fc, taskGroupStatus, exeStatus, taskGroup, allocation, wg)
		})
		if err != nil {
			log.Printf("Failed to remoteExecuteOnLocation %v: %v", allocation, err)
		}
		taskGroup.MarkStop(err)
		return err
	}

	util.ExecuteWithCleanup(
		ctx,
		func() error {
			if isRestartableTasks(tasks) {
				return util.TimeDelayedRetry(fn, time.Minute, 3*time.Minute)
			} else {
				return fn()
			}
		},
		func() {
			var w sync.WaitGroup
			for _, shard := range lastTask.OutputShards {
				w.Add(1)
				// println("deleting", shard.Name(), "from", allocation.Location.URL())
				go func(shard *flow.DatasetShard) {
					defer w.Done()
					if err := sendDeleteRequest(allocation.Location.URL(), &pb.DeleteDatasetShardRequest{
						Name: shard.Name(),
					}); err != nil {
						println("Purging dataset error:", err.Error())
					}
				}(shard)
			}
			w.Wait()
		},
	)

}
254189
func GetExecutorOutput(exec remotecommand.Executor) (*bytes.Buffer, *bytes.Buffer, error) {
	var stdOut bytes.Buffer
	var stdErr bytes.Buffer
	err := exec.Stream(remotecommand.StreamOptions{
		Stdout: &stdOut,
		Stderr: &stdErr,
		Tty:    false,
	})
	if err != nil {
		return nil, nil, errors.InternalWrapError(err)
	}
	return &stdOut, &stdErr, nil
}
1353399
func (e *Executor) StreamExecute(ctx context.Context, method string, safeSession *SafeSession, sql string, bindVars map[string]*querypb.BindVariable, target querypb.Target, callback func(*sqltypes.Result) error) (err error) {
	logStats := NewLogStats(ctx, method, sql, bindVars)
	logStats.StmtType = sqlparser.StmtType(sqlparser.Preview(sql))
	defer logStats.Send()

	if bindVars == nil {
		bindVars = make(map[string]*querypb.BindVariable)
	}
	query, comments := sqlparser.SplitMarginComments(sql)
	vcursor := newVCursorImpl(ctx, safeSession, target.Keyspace, target.TabletType, comments, e, logStats)

	// check if this is a stream statement for messaging
	// TODO: support keyRange syntax
	if logStats.StmtType == sqlparser.StmtType(sqlparser.StmtStream) {
		return e.handleMessageStream(ctx, safeSession, sql, target, callback, vcursor, logStats)
	}

	plan, err := e.getPlan(
		vcursor,
		query,
		comments,
		bindVars,
		skipQueryPlanCache(safeSession),
		logStats,
	)
	if err != nil {
		logStats.Error = err
		return err
	}

	execStart := time.Now()
	logStats.PlanTime = execStart.Sub(logStats.StartTime)

	// Some of the underlying primitives may send results one row at a time.
	// So, we need the ability to consolidate those into reasonable chunks.
	// The callback wrapper below accumulates rows and sends them as chunks
	// dictated by stream_buffer_size.
	result := &sqltypes.Result{}
	byteCount := 0
	err = plan.Instructions.StreamExecute(vcursor, bindVars, true, func(qr *sqltypes.Result) error {
		// If the row has field info, send it separately.
		// TODO(sougou): this behavior is for handling tests because
		// the framework currently sends all results as one packet.
		if len(qr.Fields) > 0 {
			qrfield := &sqltypes.Result{Fields: qr.Fields}
			if err := callback(qrfield); err != nil {
				return err
			}
		}

		for _, row := range qr.Rows {
			result.Rows = append(result.Rows, row)
			for _, col := range row {
				byteCount += col.Len()
			}

			if byteCount >= e.streamSize {
				err := callback(result)
				result = &sqltypes.Result{}
				byteCount = 0
				if err != nil {
					return err
				}
			}
		}
		return nil
	})

	// Send left-over rows.
	if len(result.Rows) > 0 {
		if err := callback(result); err != nil {
			return err
		}
	}

	logStats.ExecuteTime = time.Since(execStart)

	return err
}
298641
func (client *QueryClient) StreamExecuteWithOptions(query string, bindvars map[string]*querypb.BindVariable, options *querypb.ExecuteOptions) (*sqltypes.Result, error) {
	result := &sqltypes.Result{}
	err := client.server.StreamExecute(
		client.ctx,
		&client.target,
		query,
		bindvars,
		0,
		options,
		func(res *sqltypes.Result) error {
			if result.Fields == nil {
				result.Fields = res.Fields
			}
			result.Rows = append(result.Rows, res.Rows...)
			result.RowsAffected += uint64(len(res.Rows))
			return nil
		},
	)
	if err != nil {
		return nil, err
	}
	return result, nil
}
887053
func (me *Database) ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) {
	me.Trace("EXEC", query, args...)
	return me.Db.ExecContext(ctx, query, args...)
}
1007310
func (stmt *OCI8Stmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) {
	list := make([]namedValue, len(args))
	for i, nv := range args {
		list[i] = toNamedValue(nv)
	}
	return stmt.exec(ctx, list)
}
752933
func (s *GetMaintenanceWindowExecutionTaskOutput) SetTaskExecutionId(v string) *GetMaintenanceWindowExecutionTaskOutput {
	s.TaskExecutionId = &v
	return s
}
928949
func (p *GoTemplatePrinter) safeExecute(w io.Writer, obj interface{}) error {
	var panicErr error
	// Sorry for the double anonymous function. There's probably a clever way
	// to do this that has the defer'd func setting the value to be returned, but
	// that would be even less obvious.
	retErr := func() error {
		defer func() {
			if x := recover(); x != nil {
				panicErr = fmt.Errorf("caught panic: %+v", x)
			}
		}()
		return p.template.Execute(w, obj)
	}()
	if panicErr != nil {
		return panicErr
	}
	return retErr
}
213424
func (agent *ActionAgent) VReplicationExec(ctx context.Context, query string) (*querypb.QueryResult, error) {
	qr, err := agent.VREngine.Exec(query)
	if err != nil {
		return nil, err
	}
	return sqltypes.ResultToProto3(qr), nil
}
1099574
func (conn *gRPCQueryClient) BeginExecute(ctx context.Context, target *querypb.Target, query string, bindVars map[string]*querypb.BindVariable, options *querypb.ExecuteOptions) (result *sqltypes.Result, transactionID int64, err error) {
	conn.mu.RLock()
	defer conn.mu.RUnlock()
	if conn.cc == nil {
		return nil, 0, tabletconn.ConnClosed
	}

	req := &querypb.BeginExecuteRequest{
		Target:            target,
		EffectiveCallerId: callerid.EffectiveCallerIDFromContext(ctx),
		ImmediateCallerId: callerid.ImmediateCallerIDFromContext(ctx),
		Query: &querypb.BoundQuery{
			Sql:           query,
			BindVariables: bindVars,
		},
		Options: options,
	}
	reply, err := conn.c.BeginExecute(ctx, req)
	if err != nil {
		return nil, 0, tabletconn.ErrorFromGRPC(err)
	}
	if reply.Error != nil {
		return nil, reply.TransactionId, tabletconn.ErrorFromVTRPC(reply.Error)
	}
	return sqltypes.Proto3ToResult(reply.Result), reply.TransactionId, nil
}
1145563
func (tl *TestLauncher) MustExecuteQuery(query string) *QueryResults {
	results, err := tl.ExecuteQuery(query)
	if err != nil {
		panic(err)
	}
	return results
}
1318028
func (tsv *TabletServer) BeginExecute(ctx context.Context, target *querypb.Target, sql string, bindVariables map[string]*querypb.BindVariable, options *querypb.ExecuteOptions) (*sqltypes.Result, int64, error) {
	if tsv.enableHotRowProtection {
		txDone, err := tsv.beginWaitForSameRangeTransactions(ctx, target, options, sql, bindVariables)
		if err != nil {
			return nil, 0, err
		}
		if txDone != nil {
			defer txDone()
		}
	}

	transactionID, err := tsv.Begin(ctx, target, options)
	if err != nil {
		return nil, 0, err
	}

	result, err := tsv.Execute(ctx, target, sql, bindVariables, transactionID, options)
	return result, transactionID, err
}
767195
func (e *engineImpl) execCronTickTask(c context.Context, tqTask proto.Message) error {
	msg := tqTask.(*internal.CronTickTask)
	return e.jobTxn(c, msg.JobId, func(c context.Context, job *Job, isNew bool) error {
		if isNew {
			logging.Errorf(c, "Scheduled job is unexpectedly gone")
			return errSkipPut
		}
		logging.Infof(c, "Tick %d has arrived", msg.TickNonce)
		return pokeCron(c, job, e.cfg.Dispatcher, func(m *cron.Machine) error {
			// OnTimerTick returns an error if the tick happened too soon. Mark this
			// error as transient to trigger task queue retry at a later time.
			return transient.Tag.Apply(m.OnTimerTick(msg.TickNonce))
		})
	})
}