<commit_before>package engine

import (
	log "code.google.com/p/log4go"
	"common"
	"fmt"
	"parser"
	"protocol"
	"sort"
	"strconv"
	"strings"
	"time"
)

type QueryEngine struct {
	query          *parser.SelectQuery
	where          *parser.WhereCondition
	responseChan   chan *protocol.Response
	shouldLimit    bool
	limit          int
	seriesToPoints map[string]*protocol.Series
	yield          func(*protocol.Series) error

	// variables for aggregate queries
	isAggregateQuery    bool
	aggregators         []Aggregator
	duration            *time.Duration
	timestampAggregator Aggregator
	groups              map[string]map[Group]bool
	pointsRange         map[string]*PointRange
	groupBy             *parser.GroupByClause
	aggregateYield      func(*protocol.Series) error
}

const (
	POINT_BATCH_SIZE = 100
)

var (
	responseQuery     = protocol.Response_QUERY
	responseEndStream = protocol.Response_END_STREAM
)

// distribute query and possibly do the merge/join before yielding the points
func (self *QueryEngine) distributeQuery(query *parser.SelectQuery, yield func(*protocol.Series) error) (err error) {
	// see if this is a merge query
	fromClause := query.GetFromClause()
	if fromClause.Type == parser.FromClauseMerge {
		yield = getMergeYield(fromClause.Names[0].Name.Name, fromClause.Names[1].Name.Name, query.Ascending, yield)
	}

	if fromClause.Type == parser.FromClauseInnerJoin {
		yield = getJoinYield(query, yield)
	}

	self.yield = yield
	return nil
}

func NewQueryEngine(query *parser.SelectQuery, responseChan chan *protocol.Response) *QueryEngine {
	limit := query.Limit
	shouldLimit := true
	if limit == 0 {
		shouldLimit = false
	}

	queryEngine := &QueryEngine{
		query:          query,
		where:          query.GetWhereCondition(),
		limit:          limit,
		shouldLimit:    shouldLimit,
		responseChan:   responseChan,
		seriesToPoints: make(map[string]*protocol.Series),
	}

	yield := func(series *protocol.Series) error {
		response := &protocol.Response{Type: &responseQuery, Series: series}
		responseChan <- response
		return nil
	}

	if query.HasAggregates() {
		queryEngine.executeCountQueryWithGroupBy(query, yield)
	} else if containsArithmeticOperators(query) {
		queryEngine.executeArithmeticQuery(query, yield)
	} else {
		queryEngine.distributeQuery(query, yield)
	}

	return queryEngine
}

// Returns false if the query should be stopped (either because of limit or error)
func (self *QueryEngine) YieldPoint(seriesName *string, fieldNames []string, point *protocol.Point) (shouldContinue bool) {
	shouldContinue = true
	series := self.seriesToPoints[*seriesName]
	if series == nil {
		series = &protocol.Series{Name: protocol.String(*seriesName), Fields: fieldNames, Points: make([]*protocol.Point, 0, POINT_BATCH_SIZE)}
		self.seriesToPoints[*seriesName] = series
	} else if len(series.Points) >= POINT_BATCH_SIZE {
		shouldContinue = self.yieldSeriesData(series)
		series = &protocol.Series{Name: protocol.String(*seriesName), Fields: fieldNames, Points: make([]*protocol.Point, 0, POINT_BATCH_SIZE)}
		self.seriesToPoints[*seriesName] = series
	}
	series.Points = append(series.Points, point)

	return shouldContinue
}

func (self *QueryEngine) yieldSeriesData(series *protocol.Series) bool {
	var err error
	if self.where != nil {
		serieses, err := self.filter(series)
		if err != nil {
			log.Error("Error while filtering points: %s\n", err)
			return false
		}
		for _, series := range serieses {
			if len(series.Points) > 0 {
				self.calculateLimitAndSlicePoints(series)
				if len(series.Points) > 0 {
					err = self.yield(series)
				}
			}
		}
	} else {
		self.calculateLimitAndSlicePoints(series)

		if len(series.Points) > 0 {
			err = self.yield(series)
		}
	}
	if err != nil {
		return false
	}
	return !self.hitLimit()
}

// TODO: make limits work for aggregate queries and for queries that pull from multiple series.
func (self *QueryEngine) calculateLimitAndSlicePoints(series *protocol.Series) {
	if !self.isAggregateQuery && self.shouldLimit {
		self.limit -= len(series.Points)
		if self.limit < 0 {
			sliceTo := len(series.Points) + self.limit
			if sliceTo > 0 {
				series.Points = series.Points[0:sliceTo]
			}
		}
	}
}

func (self *QueryEngine) hitLimit() bool {
	if self.isAggregateQuery || !self.shouldLimit {
		return false
	}
	return self.limit < 1
}

func (self *QueryEngine) filter(series *protocol.Series) ([]*protocol.Series, error) {
	aliases := self.query.GetTableAliases(*series.Name)
	result := make([]*protocol.Series, len(aliases), len(aliases))
	for i, alias := range aliases {
		_alias := alias
		newSeries := &protocol.Series{Name: &_alias, Points: series.Points, Fields: series.Fields}

		filteredResult := newSeries
		var err error

		// var err error
		if self.query.GetFromClause().Type != parser.FromClauseInnerJoin {
			filteredResult, err = Filter(self.query, newSeries)
			if err != nil {
				return nil, err
			}
		}
		result[i] = filteredResult
	}
	return result, nil
}

func (self *QueryEngine) Close() {
	for _, series := range self.seriesToPoints {
		if len(series.Points) == 0 {
			continue
		}
		self.yieldSeriesData(series)
	}

	var err error
	for _, series := range self.seriesToPoints {
		s := &protocol.Series{
			Name:   series.Name,
			Fields: series.Fields,
		}
		if len(s.Points) > 0 {
			err = self.yield(s)
			if err != nil {
				break
			}
		}
	}

	if self.isAggregateQuery {
		self.runAggregates()
	}
	response := &protocol.Response{Type: &responseEndStream}
	if err != nil {
		message := err.Error()
		response.ErrorMessage = &message
	}
	self.responseChan <- response
}

func containsArithmeticOperators(query *parser.SelectQuery) bool {
	for _, column := range query.GetColumnNames() {
		if column.Type == parser.ValueExpression {
			return true
		}
	}
	return false
}

func getTimestampFromPoint(window time.Duration, point *protocol.Point) int64 {
	multiplier := int64(window / time.Microsecond)
	return *point.GetTimestampInMicroseconds() / int64(multiplier) * int64(multiplier)
}
<commit_msg>remove the if statement to actually yield empty series<commit_after>package engine

import (
	log "code.google.com/p/log4go"
	"common"
	"fmt"
	"parser"
	"protocol"
	"sort"
	"strconv"
	"strings"
	"time"
)

type QueryEngine struct {
	query          *parser.SelectQuery
	where          *parser.WhereCondition
	responseChan   chan *protocol.Response
	shouldLimit    bool
	limit          int
	seriesToPoints map[string]*protocol.Series
	yield          func(*protocol.Series) error

	// variables for aggregate queries
	isAggregateQuery    bool
	aggregators         []Aggregator
	duration            *time.Duration
	timestampAggregator Aggregator
	groups              map[string]map[Group]bool
	pointsRange         map[string]*PointRange
	groupBy             *parser.GroupByClause
	aggregateYield      func(*protocol.Series) error
}

const (
	POINT_BATCH_SIZE = 100
)

var (
	responseQuery     = protocol.Response_QUERY
	responseEndStream = protocol.Response_END_STREAM
)

// distribute query and possibly do the merge/join before yielding the points
func (self *QueryEngine) distributeQuery(query *parser.SelectQuery, yield func(*protocol.Series) error) (err error) {
	// see if this is a merge query
	fromClause := query.GetFromClause()
	if fromClause.Type == parser.FromClauseMerge {
		yield = getMergeYield(fromClause.Names[0].Name.Name, fromClause.Names[1].Name.Name, query.Ascending, yield)
	}

	if fromClause.Type == parser.FromClauseInnerJoin {
		yield = getJoinYield(query, yield)
	}

	self.yield = yield
	return nil
}

func NewQueryEngine(query *parser.SelectQuery, responseChan chan *protocol.Response) *QueryEngine {
	limit := query.Limit
	shouldLimit := true
	if limit == 0 {
		shouldLimit = false
	}

	queryEngine := &QueryEngine{
		query:          query,
		where:          query.GetWhereCondition(),
		limit:          limit,
		shouldLimit:    shouldLimit,
		responseChan:   responseChan,
		seriesToPoints: make(map[string]*protocol.Series),
	}

	yield := func(series *protocol.Series) error {
		response := &protocol.Response{Type: &responseQuery, Series: series}
		responseChan <- response
		return nil
	}

	if query.HasAggregates() {
		queryEngine.executeCountQueryWithGroupBy(query, yield)
	} else if containsArithmeticOperators(query) {
		queryEngine.executeArithmeticQuery(query, yield)
	} else {
		queryEngine.distributeQuery(query, yield)
	}

	return queryEngine
}

// Returns false if the query should be stopped (either because of limit or error)
func (self *QueryEngine) YieldPoint(seriesName *string, fieldNames []string, point *protocol.Point) (shouldContinue bool) {
	shouldContinue = true
	series := self.seriesToPoints[*seriesName]
	if series == nil {
		series = &protocol.Series{Name: protocol.String(*seriesName), Fields: fieldNames, Points: make([]*protocol.Point, 0, POINT_BATCH_SIZE)}
		self.seriesToPoints[*seriesName] = series
	} else if len(series.Points) >= POINT_BATCH_SIZE {
		shouldContinue = self.yieldSeriesData(series)
		series = &protocol.Series{Name: protocol.String(*seriesName), Fields: fieldNames, Points: make([]*protocol.Point, 0, POINT_BATCH_SIZE)}
		self.seriesToPoints[*seriesName] = series
	}
	series.Points = append(series.Points, point)

	return shouldContinue
}

func (self *QueryEngine) yieldSeriesData(series *protocol.Series) bool {
	var err error
	if self.where != nil {
		serieses, err := self.filter(series)
		if err != nil {
			log.Error("Error while filtering points: %s\n", err)
			return false
		}
		for _, series := range serieses {
			if len(series.Points) > 0 {
				self.calculateLimitAndSlicePoints(series)
				if len(series.Points) > 0 {
					err = self.yield(series)
				}
			}
		}
	} else {
		self.calculateLimitAndSlicePoints(series)

		if len(series.Points) > 0 {
			err = self.yield(series)
		}
	}
	if err != nil {
		return false
	}
	return !self.hitLimit()
}

// TODO: make limits work for aggregate queries and for queries that pull from multiple series.
func (self *QueryEngine) calculateLimitAndSlicePoints(series *protocol.Series) {
	if !self.isAggregateQuery && self.shouldLimit {
		self.limit -= len(series.Points)
		if self.limit < 0 {
			sliceTo := len(series.Points) + self.limit
			if sliceTo > 0 {
				series.Points = series.Points[0:sliceTo]
			}
		}
	}
}

func (self *QueryEngine) hitLimit() bool {
	if self.isAggregateQuery || !self.shouldLimit {
		return false
	}
	return self.limit < 1
}

func (self *QueryEngine) filter(series *protocol.Series) ([]*protocol.Series, error) {
	aliases := self.query.GetTableAliases(*series.Name)
	result := make([]*protocol.Series, len(aliases), len(aliases))
	for i, alias := range aliases {
		_alias := alias
		newSeries := &protocol.Series{Name: &_alias, Points: series.Points, Fields: series.Fields}

		filteredResult := newSeries
		var err error

		// var err error
		if self.query.GetFromClause().Type != parser.FromClauseInnerJoin {
			filteredResult, err = Filter(self.query, newSeries)
			if err != nil {
				return nil, err
			}
		}
		result[i] = filteredResult
	}
	return result, nil
}

func (self *QueryEngine) Close() {
	for _, series := range self.seriesToPoints {
		if len(series.Points) == 0 {
			continue
		}
		self.yieldSeriesData(series)
	}

	var err error
	for _, series := range self.seriesToPoints {
		s := &protocol.Series{
			Name:   series.Name,
			Fields: series.Fields,
		}
		err = self.yield(s)
		if err != nil {
			break
		}
	}

	if self.isAggregateQuery {
		self.runAggregates()
	}
	response := &protocol.Response{Type: &responseEndStream}
	if err != nil {
		message := err.Error()
		response.ErrorMessage = &message
	}
	self.responseChan <- response
}

func containsArithmeticOperators(query *parser.SelectQuery) bool {
	for _, column := range query.GetColumnNames() {
		if column.Type == parser.ValueExpression {
			return true
		}
	}
	return false
}

func getTimestampFromPoint(window time.Duration, point *protocol.Point) int64 {
	multiplier := int64(window / time.Microsecond)
	return *point.GetTimestampInMicroseconds() / int64(multiplier) * int64(multiplier)
}

{'fraction_non_alphanumeric': 0.1126807379092571, 'fraction_numerical': 0.004237992354994183, 'mean_word_length': 3.928337428337428, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6397385', 'n_tokens_mistral': 4129, 'n_tokens_neox': 3807, 'n_words': 933}