func (q *QueryTask) close() {
	q.mu.Lock()
	if q.status != KilledTask {
		q.status = KilledTask
		close(q.closing)
	}
	q.mu.Unlock()
}
close closes the query task closing channel if the query hasn't been previously killed.
package query
import (
	"context"
	"errors"
	"fmt"
	"os"
	"runtime/debug"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxql"
	"go.uber.org/zap"
)
var (
	ErrInvalidQuery = errors.New("invalid query")
	ErrNotExecuted = errors.New("not executed")
	ErrQueryInterrupted = errors.New("query interrupted")
	ErrQueryAborted = errors.New("query aborted")
	ErrQueryEngineShutdown = errors.New("query engine shutdown")
	ErrQueryTimeoutLimitExceeded = errors.New("query-timeout limit exceeded")
	ErrAlreadyKilled = errors.New("already killed")
)
const (
	statQueriesActive          = "queriesActive"    
	statQueriesExecuted        = "queriesExecuted"  
	statQueriesFinished        = "queriesFinished"  
	statQueryExecutionDuration = "queryDurationNs"  
	statRecoveredPanics        = "recoveredPanics"  
	PanicCrashEnv = "INFLUXDB_PANIC_CRASH"
)
func ErrDatabaseNotFound(name string) error { return fmt.Errorf("database not found: %s", name) }
func ErrMaxSelectPointsLimitExceeded(n, limit int) error {
	return fmt.Errorf("max-select-point limit exceeed: (%d/%d)", n, limit)
}
func ErrMaxConcurrentQueriesLimitExceeded(n, limit int) error {
	return fmt.Errorf("max-concurrent-queries limit exceeded(%d, %d)", n, limit)
}
type Authorizer interface {
	AuthorizeDatabase(p influxql.Privilege, name string) bool
	AuthorizeQuery(database string, query *influxql.Query) error
	AuthorizeSeriesRead(database string, measurement []byte, tags models.Tags) bool
	AuthorizeSeriesWrite(database string, measurement []byte, tags models.Tags) bool
}
type openAuthorizer struct{}
var OpenAuthorizer = openAuthorizer{}
func (a openAuthorizer) AuthorizeDatabase(influxql.Privilege, string) bool { return true }
func (a openAuthorizer) AuthorizeSeriesRead(database string, measurement []byte, tags models.Tags) bool {
	return true
}
func (a openAuthorizer) AuthorizeSeriesWrite(database string, measurement []byte, tags models.Tags) bool {
	return true
}
func (a openAuthorizer) AuthorizeQuery(_ string, _ *influxql.Query) error { return nil }
func AuthorizerIsOpen(a Authorizer) bool {
	return a == nil || a == OpenAuthorizer
}
type ExecutionOptions struct {
	Database string
	Authorizer Authorizer
	ChunkSize int
	ReadOnly bool
	NodeID uint64
	Quiet bool
	AbortCh <-chan struct{}
}
type ExecutionContext struct {
	StatementID int
	QueryID uint64
	Query *QueryTask
	Results chan *Result
	InterruptCh <-chan struct{}
	ExecutionOptions
}
func (ctx *ExecutionContext) send(result *Result) error {
	select {
	case <-ctx.AbortCh:
		return ErrQueryAborted
	case ctx.Results <- result:
	}
	return nil
}
func (ctx *ExecutionContext) Send(result *Result) error {
	select {
	case <-ctx.InterruptCh:
		return ErrQueryInterrupted
	case <-ctx.AbortCh:
		return ErrQueryAborted
	case ctx.Results <- result:
	}
	return nil
}
type contextKey int
const (
	iteratorsContextKey contextKey = iota
)
func NewContextWithIterators(ctx context.Context, itr *Iterators) context.Context {
	return context.WithValue(ctx, iteratorsContextKey, itr)
}
func tryAddAuxIteratorToContext(ctx context.Context, itr AuxIterator) {
	if v, ok := ctx.Value(iteratorsContextKey).(*Iterators); ok {
		*v = append(*v, itr)
	}
}
type StatementExecutor interface {
	ExecuteStatement(stmt influxql.Statement, ctx ExecutionContext) error
}
type StatementNormalizer interface {
	NormalizeStatement(stmt influxql.Statement, database string) error
}
type QueryExecutor struct {
	StatementExecutor StatementExecutor
	TaskManager *TaskManager
	Logger *zap.Logger
	stats *QueryStatistics
}
func NewQueryExecutor() *QueryExecutor {
	return &QueryExecutor{
		TaskManager: NewTaskManager(),
		Logger:      zap.NewNop(),
		stats:       &QueryStatistics{},
	}
}
type QueryStatistics struct {
	ActiveQueries          int64
	ExecutedQueries        int64
	FinishedQueries        int64
	QueryExecutionDuration int64
	RecoveredPanics        int64
}
func (e *QueryExecutor) Statistics(tags map[string]string) []models.Statistic {
	return []models.Statistic{{
		Name: "queryExecutor",
		Tags: tags,
		Values: map[string]interface{}{
			statQueriesActive:          atomic.LoadInt64(&e.stats.ActiveQueries),
			statQueriesExecuted:        atomic.LoadInt64(&e.stats.ExecutedQueries),
			statQueriesFinished:        atomic.LoadInt64(&e.stats.FinishedQueries),
			statQueryExecutionDuration: atomic.LoadInt64(&e.stats.QueryExecutionDuration),
			statRecoveredPanics:        atomic.LoadInt64(&e.stats.RecoveredPanics),
		},
	}}
}
func (e *QueryExecutor) Close() error {
	return e.TaskManager.Close()
}
func (e *QueryExecutor) WithLogger(log *zap.Logger) {
	e.Logger = log.With(zap.String("service", "query"))
	e.TaskManager.Logger = e.Logger
}
func (e *QueryExecutor) ExecuteQuery(query *influxql.Query, opt ExecutionOptions, closing chan struct{}) <-chan *Result {
	results := make(chan *Result)
	go e.executeQuery(query, opt, closing, results)
	return results
}
func (e *QueryExecutor) executeQuery(query *influxql.Query, opt ExecutionOptions, closing <-chan struct{}, results chan *Result) {
	defer close(results)
	defer e.recover(query, results)
	atomic.AddInt64(&e.stats.ActiveQueries, 1)
	atomic.AddInt64(&e.stats.ExecutedQueries, 1)
	defer func(start time.Time) {
		atomic.AddInt64(&e.stats.ActiveQueries, -1)
		atomic.AddInt64(&e.stats.FinishedQueries, 1)
		atomic.AddInt64(&e.stats.QueryExecutionDuration, time.Since(start).Nanoseconds())
	}(time.Now())
	qid, task, err := e.TaskManager.AttachQuery(query, opt.Database, closing)
	if err != nil {
		select {
		case results <- &Result{Err: err}:
		case <-opt.AbortCh:
		}
		return
	}
	defer e.TaskManager.DetachQuery(qid)
	ctx := ExecutionContext{
		QueryID:          qid,
		Query:            task,
		Results:          results,
		InterruptCh:      task.closing,
		ExecutionOptions: opt,
	}
	var i int
LOOP:
	for ; i < len(query.Statements); i++ {
		ctx.StatementID = i
		stmt := query.Statements[i]
		defaultDB := opt.Database
		if defaultDB == "" {
			if s, ok := stmt.(influxql.HasDefaultDatabase); ok {
				defaultDB = s.DefaultDatabase()
			}
		}
		if stmt, ok := stmt.(*influxql.SelectStatement); ok {
			for _, s := range stmt.Sources {
				switch s := s.(type) {
				case *influxql.Measurement:
					if influxql.IsSystemName(s.Name) {
						command := "the appropriate meta command"
						switch s.Name {
						case "_fieldKeys":
							command = "SHOW FIELD KEYS"
						case "_measurements":
							command = "SHOW MEASUREMENTS"
						case "_series":
							command = "SHOW SERIES"
						case "_tagKeys":
							command = "SHOW TAG KEYS"
						case "_tags":
							command = "SHOW TAG VALUES"
						}
						results <- &Result{
							Err: fmt.Errorf("unable to use system source '%s': use %s instead", s.Name, command),
						}
						break LOOP
					}
				}
			}
		}
		newStmt, err := RewriteStatement(stmt)
		if err != nil {
			results <- &Result{Err: err}
			break
		}
		stmt = newStmt
		if normalizer, ok := e.StatementExecutor.(StatementNormalizer); ok {
			if err := normalizer.NormalizeStatement(stmt, defaultDB); err != nil {
				if err := ctx.send(&Result{Err: err}); err == ErrQueryAborted {
					return
				}
				break
			}
		}
		if !ctx.Quiet {
			e.Logger.Info("Executing query", zap.Stringer("query", stmt))
		}
		err = e.StatementExecutor.ExecuteStatement(stmt, ctx)
		if err == ErrQueryInterrupted {
			if qerr := task.Error(); qerr != nil {
				err = qerr
			}
		}
		if err != nil {
			if err := ctx.send(&Result{
				StatementID: i,
				Err:         err,
			}); err == ErrQueryAborted {
				return
			}
			break
		}
		interrupted := false
		if ctx.InterruptCh != nil {
			select {
			case <-ctx.InterruptCh:
				interrupted = true
			default:
			}
		}
		if interrupted {
			break
		}
	}
	for ; i < len(query.Statements)-1; i++ {
		if err := ctx.send(&Result{
			StatementID: i,
			Err:         ErrNotExecuted,
		}); err == ErrQueryAborted {
			return
		}
	}
}
var willCrash bool
func init() {
	var err error
	if willCrash, err = strconv.ParseBool(os.Getenv(PanicCrashEnv)); err != nil {
		willCrash = false
	}
}
func (e *QueryExecutor) recover(query *influxql.Query, results chan *Result) {
	if err := recover(); err != nil {
		atomic.AddInt64(&e.stats.RecoveredPanics, 1)  
		e.Logger.Error(fmt.Sprintf("%s [panic:%s] %s", query.String(), err, debug.Stack()))
		results <- &Result{
			StatementID: -1,
			Err:         fmt.Errorf("%s [panic:%s]", query.String(), err),
		}
		if willCrash {
			e.Logger.Error(fmt.Sprintf("\n\n=====\nAll goroutines now follow:"))
			buf := debug.Stack()
			e.Logger.Error(fmt.Sprintf("%s", buf))
			os.Exit(1)
		}
	}
}
type QueryMonitorFunc func(<-chan struct{}) error
type QueryTask struct {
	query     string
	database  string
	status    TaskStatus
	startTime time.Time
	closing   chan struct{}
	monitorCh chan error
	err       error
	mu        sync.Mutex
}
func (q *QueryTask) Monitor(fn QueryMonitorFunc) {
	go q.monitor(fn)
}
func (q *QueryTask) Error() error {
	q.mu.Lock()
	defer q.mu.Unlock()
	return q.err
}
func (q *QueryTask) setError(err error) {
	q.mu.Lock()
	q.err = err
	q.mu.Unlock()
}
func (q *QueryTask) monitor(fn QueryMonitorFunc) {
	if err := fn(q.closing); err != nil {
		select {
		case <-q.closing:
		case q.monitorCh <- err:
		}
	}
}