func (t *TaskManager) AttachQuery(q *influxql.Query, database string, interrupt <-chan struct{}) (uint64, *QueryTask, error) {
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.shutdown {
		return 0, nil, ErrQueryEngineShutdown
	}
	if t.MaxConcurrentQueries > 0 && len(t.queries) >= t.MaxConcurrentQueries {
		return 0, nil, ErrMaxConcurrentQueriesLimitExceeded(len(t.queries), t.MaxConcurrentQueries)
	}
	qid := t.nextID
	query := &QueryTask{
		query:     q.String(),
		database:  database,
		status:    RunningTask,
		startTime: time.Now(),
		closing:   make(chan struct{}),
		monitorCh: make(chan error),
	}
	t.queries[qid] = query
	go t.waitForQuery(qid, query.closing, interrupt, query.monitorCh)
	if t.LogQueriesAfter != 0 {
		go query.monitor(func(closing <-chan struct{}) error {
			timer := time.NewTimer(t.LogQueriesAfter)
			defer timer.Stop()
			select {
			case <-timer.C:
				t.Logger.Warn(fmt.Sprintf("Detected slow query: %s (qid: %d, database: %s, threshold: %s)",
					query.query, qid, query.database, t.LogQueriesAfter))
			case <-closing:
			}
			return nil
		})
	}
	t.nextID++
	return qid, query, nil
}
AttachQuery attaches a running query to be managed by the TaskManager.
Returns the query id of the newly attached query or an error if it was
unable to assign a query id or attach the query to the TaskManager.
This function also returns a channel that will be closed when this
query finishes running.

After a query finishes running, the system is free to reuse a query id.
package query
import (
	"bytes"
	"encoding/json"
	"fmt"
	"sync"
	"time"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxql"
	"go.uber.org/zap"
)
const (
	DefaultQueryTimeout = time.Duration(0)
)
type TaskStatus int
const (
	RunningTask TaskStatus = iota + 1
	KilledTask
)
func (t TaskStatus) String() string {
	switch t {
	case RunningTask:
		return "running"
	case KilledTask:
		return "killed"
	default:
		return "unknown"
	}
}
func (t TaskStatus) MarshalJSON() ([]byte, error) {
	s := t.String()
	return json.Marshal(s)
}
func (t *TaskStatus) UnmarshalJSON(data []byte) error {
	if bytes.Equal(data, []byte("running")) {
		*t = RunningTask
	} else if bytes.Equal(data, []byte("killed")) {
		*t = KilledTask
	} else if bytes.Equal(data, []byte("unknown")) {
		*t = TaskStatus(0)
	} else {
		return fmt.Errorf("unknown task status: %s", string(data))
	}
	return nil
}
type TaskManager struct {
	QueryTimeout time.Duration
	LogQueriesAfter time.Duration
	MaxConcurrentQueries int
	Logger *zap.Logger
	queries  map[uint64]*QueryTask
	nextID   uint64
	mu       sync.RWMutex
	shutdown bool
}
func NewTaskManager() *TaskManager {
	return &TaskManager{
		QueryTimeout: DefaultQueryTimeout,
		Logger:       zap.NewNop(),
		queries:      make(map[uint64]*QueryTask),
		nextID:       1,
	}
}
func (t *TaskManager) ExecuteStatement(stmt influxql.Statement, ctx ExecutionContext) error {
	switch stmt := stmt.(type) {
	case *influxql.ShowQueriesStatement:
		rows, err := t.executeShowQueriesStatement(stmt)
		if err != nil {
			return err
		}
		ctx.Results <- &Result{
			StatementID: ctx.StatementID,
			Series:      rows,
		}
	case *influxql.KillQueryStatement:
		var messages []*Message
		if ctx.ReadOnly {
			messages = append(messages, ReadOnlyWarning(stmt.String()))
		}
		if err := t.executeKillQueryStatement(stmt); err != nil {
			return err
		}
		ctx.Results <- &Result{
			StatementID: ctx.StatementID,
			Messages:    messages,
		}
	default:
		return ErrInvalidQuery
	}
	return nil
}
func (t *TaskManager) executeKillQueryStatement(stmt *influxql.KillQueryStatement) error {
	return t.KillQuery(stmt.QueryID)
}
func (t *TaskManager) executeShowQueriesStatement(q *influxql.ShowQueriesStatement) (models.Rows, error) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	now := time.Now()
	values := make([][]interface{}, 0, len(t.queries))
	for id, qi := range t.queries {
		d := now.Sub(qi.startTime)
		switch {
		case d >= time.Second:
			d = d - (d % time.Second)
		case d >= time.Millisecond:
			d = d - (d % time.Millisecond)
		case d >= time.Microsecond:
			d = d - (d % time.Microsecond)
		}
		values = append(values, []interface{}{id, qi.query, qi.database, d.String(), qi.status.String()})
	}
	return []*models.Row{{
		Columns: []string{"qid", "query", "database", "duration", "status"},
		Values:  values,
	}}, nil
}
func (t *TaskManager) queryError(qid uint64, err error) {
	t.mu.RLock()
	query := t.queries[qid]
	t.mu.RUnlock()
	if query != nil {
		query.setError(err)
	}
}