<commit_before>package engine

import (
	"common"
	"coordinator"
	"fmt"
	"os"
	"parser"
	"protocol"
	"runtime"
	"time"
)

type QueryEngine struct {
	coordinator coordinator.Coordinator
}

func (self *QueryEngine) RunQuery(query *parser.Query, yield func(*protocol.Series) error) (err error) {
	// don't let a panic pass beyond RunQuery
	defer func() {
		if err := recover(); err != nil {
			fmt.Fprintf(os.Stderr, "********************************BUG********************************\n")
			buf := make([]byte, 1024)
			n := runtime.Stack(buf, false)
			fmt.Fprintf(os.Stderr, "Stacktrace: %s\n", string(buf[:n]))
			err = common.NewQueryError(common.InternalError, "Internal Error")
		}
	}()

	if isCountQuery(query) {
		if groupBy := query.GetGroupByClause(); len(groupBy) > 0 {
			return self.executeCountQueryWithGroupBy(query, yield)
		} else {
			return self.executeCountQuery(query, yield)
		}
	} else if isFunctionQuery(query) {
		if groupBy := query.GetGroupByClause(); len(groupBy) > 0 {
			return self.executeFunctionQueryWithGroupBy(query, yield)
		} else {
			// return self.executeMinQuery(query, yield)
		}
	} else {
		self.coordinator.DistributeQuery(query, yield)
	}
	return nil
}

func NewQueryEngine(c coordinator.Coordinator) (EngineI, error) {
	return &QueryEngine{c}, nil
}

func isCountQuery(query *parser.Query) bool {
	for _, column := range query.GetColumnNames() {
		if column.IsFunctionCall() && column.Name == "count" {
			return true
		}
	}

	return false
}

func isFunctionQuery(query *parser.Query) bool {
	for _, column := range query.GetColumnNames() {
		if column.IsFunctionCall() && (column.Name == "min" || column.Name == "max") {
			return true
		}
	}

	return false
}

func (self *QueryEngine) executeCountQuery(query *parser.Query, yield func(*protocol.Series) error) error {
	count := make(map[string]int32)
	var timestamp int64 = 0

	self.coordinator.DistributeQuery(query, func(series *protocol.Series) error {
		c := count[*series.Name]
		c += int32(len(series.Points))
		count[*series.Name] = c
		if len(series.Points) > 0 {
			timestamp = series.Points[0].GetTimestamp()
		}
		return nil
	})

	expectedFieldType := protocol.FieldDefinition_INT32
	expectedName := "count"
	var sequenceNumber uint32 = 1

	for name, value := range count {
		tempValue := value
		tempName := name

		expectedData := &protocol.Series{
			Name: &tempName,
			Fields: []*protocol.FieldDefinition{
				&protocol.FieldDefinition{Name: &expectedName, Type: &expectedFieldType},
			},
			Points: []*protocol.Point{
				&protocol.Point{
					Timestamp:      &timestamp,
					SequenceNumber: &sequenceNumber,
					Values: []*protocol.FieldValue{
						&protocol.FieldValue{
							IntValue: &tempValue,
						},
					},
				},
			},
		}
		yield(expectedData)
	}
	return nil
}

func getValueFromPoint(value *protocol.FieldValue, fType protocol.FieldDefinition_Type) interface{} {
	switch fType {
	case protocol.FieldDefinition_STRING:
		return *value.StringValue
	case protocol.FieldDefinition_INT32:
		return *value.IntValue
	case protocol.FieldDefinition_BOOL:
		return *value.BoolValue
	case protocol.FieldDefinition_DOUBLE:
		return *value.DoubleValue
	default:
		panic("WTF")
	}
}

func getTimestampFromPoint(window time.Duration, point *protocol.Point) int64 {
	return time.Unix(*point.Timestamp, 0).Round(window).Unix()
}

type Mapper func(*protocol.Point) interface{}
type InverseMapper func(interface{}, int) interface{}

// Returns a mapper and inverse mapper. A mapper is a function to map
// a point to a group and return an identifier of the group that can
// be used in a map (i.e. the returned interface must be hashable).
// An inverse mapper, takes a result of the mapper identifier and
// return the column values and/or timestamp bucket that defines the
// given group.
func createValuesToInterface(groupBy parser.GroupByClause, definitions []*protocol.FieldDefinition) (Mapper, InverseMapper, error) {
	// we shouldn't get an error, this is checked earlier in the executeCountQueryWithGroupBy
	window, _ := groupBy.GetGroupByTime()
	names := []string{}
	for _, value := range groupBy {
		if value.IsFunctionCall() {
			continue
		}
		names = append(names, value.Name)
	}

	switch len(names) {
	case 0:
		// this must be group by time
		return func(p *protocol.Point) interface{} {
				return getTimestampFromPoint(*window, p)
			}, func(i interface{}, idx int) interface{} {
				return i
			}, nil

	case 1:
		// otherwise, find the type of the column and create a mapper
		idx := -1
		var fType protocol.FieldDefinition_Type
		for index, definition := range definitions {
			if *definition.Name == names[0] {
				idx = index
				fType = *definition.Type
				break
			}
		}

		if idx == -1 {
			return nil, nil, common.NewQueryError(common.InvalidArgument, fmt.Sprintf("Invalid column name %s", groupBy[0].Name))
		}

		if window != nil {
			return func(p *protocol.Point) interface{} {
					return [2]interface{}{
						getTimestampFromPoint(*window, p),
						getValueFromPoint(p.Values[idx], fType),
					}
				}, func(i interface{}, idx int) interface{} {
					return i.([2]interface{})[idx]
				}, nil
		}
		return func(p *protocol.Point) interface{} {
				return getValueFromPoint(p.Values[idx], fType)
			}, func(i interface{}, idx int) interface{} {
				return i
			}, nil

	case 2:
		names := []string{}
		for _, value := range groupBy {
			names = append(names, value.Name)
		}

		idx1, idx2 := -1, -1
		var fType1, fType2 protocol.FieldDefinition_Type

		for index, definition := range definitions {
			if *definition.Name == names[0] {
				idx1 = index
				fType1 = *definition.Type
			} else if *definition.Name == names[1] {
				idx2 = index
				fType2 = *definition.Type
			}
			if idx1 > 0 && idx2 > 0 {
				break
			}
		}

		if window != nil {
			return func(p *protocol.Point) interface{} {
					return [3]interface{}{
						getTimestampFromPoint(*window, p),
						getValueFromPoint(p.Values[idx1], fType1),
						getValueFromPoint(p.Values[idx2], fType2),
					}
				}, func(i interface{}, idx int) interface{} {
					return i.([3]interface{})[idx]
				}, nil
		}

		return func(p *protocol.Point) interface{} {
				return [2]interface{}{
					getValueFromPoint(p.Values[idx1], fType1),
					getValueFromPoint(p.Values[idx2], fType2),
				}
			}, func(i interface{}, idx int) interface{} {
				return i.([2]interface{})[idx]
			}, nil

	default:
		// TODO: return an error instead of killing the entire process
		return nil, nil, common.NewQueryError(common.InvalidArgument, "Group by currently support up to two columns and an optional group by time")
	}
}

func (self *QueryEngine) executeCountQueryWithGroupBy(query *parser.Query, yield func(*protocol.Series) error) error {
	counts := make(map[interface{}]int32)
	timestamps := make(map[interface{}]int64)

	groupBy := query.GetGroupByClause()

	fieldTypes := map[string]*protocol.FieldDefinition_Type{}
	var inverse InverseMapper

	duration, err := groupBy.GetGroupByTime()
	if err != nil {
		return err
	}

	err = self.coordinator.DistributeQuery(query, func(series *protocol.Series) error {
		var mapper Mapper
		mapper, inverse, err = createValuesToInterface(groupBy, series.Fields)
		if err != nil {
			return err
		}

		for _, field := range series.Fields {
			fieldTypes[*field.Name] = field.Type
		}

		for _, point := range series.Points {
			value := mapper(point)
			c := counts[value]
			counts[value] = c + 1

			if duration != nil {
				timestamps[value] = getTimestampFromPoint(*duration, point)
			} else {
				timestamps[value] = point.GetTimestamp()
			}
		}

		return nil
	})

	if err != nil {
		return err
	}

	expectedFieldType := protocol.FieldDefinition_INT32
	expectedName := "count"
	var sequenceNumber uint32 = 1

	/* fields := []*protocol.FieldDefinition{} */
	points := []*protocol.Point{}

	fields := []*protocol.FieldDefinition{
		&protocol.FieldDefinition{Name: &expectedName, Type: &expectedFieldType},
	}

	for _, value := range groupBy {
		if value.IsFunctionCall() {
			continue
		}

		tempName := value.Name
		fields = append(fields, &protocol.FieldDefinition{Name: &tempName, Type: fieldTypes[tempName]})
	}

	for key, count := range counts {
		tempKey := key
		tempCount := count

		timestamp := timestamps[tempKey]

		point := &protocol.Point{
			Timestamp:      &timestamp,
			SequenceNumber: &sequenceNumber,
			Values: []*protocol.FieldValue{
				&protocol.FieldValue{
					IntValue: &tempCount,
				},
			},
		}

		for idx, _ := range groupBy {
			value := inverse(tempKey, idx)

			switch x := value.(type) {
			case string:
				point.Values = append(point.Values, &protocol.FieldValue{StringValue: &x})
			case int32:
				point.Values = append(point.Values, &protocol.FieldValue{IntValue: &x})
			case bool:
				point.Values = append(point.Values, &protocol.FieldValue{BoolValue: &x})
			case float64:
				point.Values = append(point.Values, &protocol.FieldValue{DoubleValue: &x})
			}
		}

		points = append(points, point)
	}

	expectedData := &protocol.Series{
		Name:   &query.GetFromClause().Name,
		Fields: fields,
		Points: points,
	}
	yield(expectedData)
	return nil
}

func (self *QueryEngine) executeFunctionQueryWithGroupBy(query *parser.Query, yield func(*protocol.Series) error) error {
	results := make(map[interface{}]int32)
	timestamps := make(map[interface{}]int64)
	var functionType string
	var functionField string
	var inverse InverseMapper

	groupBy := query.GetGroupByClause()

	for _, column := range query.GetColumnNames() {
		if column.IsFunctionCall() && (column.Name == "min" || column.Name == "max") {
			functionType = column.Name
			functionField = column.Elems[0].Name
		}
	}

	fieldTypes := map[string]*protocol.FieldDefinition_Type{}

	duration, err := groupBy.GetGroupByTime()
	if err != nil {
		return err
	}

	self.coordinator.DistributeQuery(query, func(series *protocol.Series) error {
		var mapper Mapper
		mapper, inverse, err = createValuesToInterface(groupBy, series.Fields)
		var fieldIndex int

		for idx, field := range series.Fields {
			fieldTypes[*field.Name] = field.Type
			if *field.Name == functionField {
				fieldIndex = idx
			}
		}

		for _, point := range series.Points {
			key := mapper(point)

			if functionType == "min" {
				min := *point.Values[fieldIndex].IntValue

				if oldMin, exists := results[key]; exists {
					if min < oldMin {
						results[key] = min
					}
				} else {
					results[key] = min
				}
			}

			if functionType == "max" {
				max := *point.Values[fieldIndex].IntValue

				if oldMax, exists := results[key]; exists {
					if max > oldMax {
						results[key] = max
					}
				} else {
					results[key] = max
				}
			}

			if duration != nil {
				timestamps[key] = getTimestampFromPoint(*duration, point)
			} else {
				timestamps[key] = point.GetTimestamp()
			}
		}

		return nil
	})

	expectedFieldType := protocol.FieldDefinition_INT32
	expectedName := functionType
	var sequenceNumber uint32 = 1

	points := []*protocol.Point{}

	fields := []*protocol.FieldDefinition{
		&protocol.FieldDefinition{Name: &expectedName, Type: &expectedFieldType},
	}

	for _, value := range groupBy {
		if value.IsFunctionCall() {
			continue
		}

		tempName := value.Name
		fields = append(fields, &protocol.FieldDefinition{Name: &tempName, Type: fieldTypes[tempName]})
	}

	for key, result := range results {
		tempKey := key
		tempValue := result

		timestamp := timestamps[tempKey]

		point := &protocol.Point{
			Timestamp:      &timestamp,
			SequenceNumber: &sequenceNumber,
			Values: []*protocol.FieldValue{
				&protocol.FieldValue{
					IntValue: &tempValue,
				},
			},
		}

		for idx, _ := range groupBy {
			value := inverse(tempKey, idx)

			switch x := value.(type) {
			case string:
				point.Values = append(point.Values, &protocol.FieldValue{StringValue: &x})
			case int32:
				point.Values = append(point.Values, &protocol.FieldValue{IntValue: &x})
			case bool:
				point.Values = append(point.Values, &protocol.FieldValue{BoolValue: &x})
			case float64:
				point.Values = append(point.Values, &protocol.FieldValue{DoubleValue: &x})
			}
		}

		points = append(points, point)
	}

	expectedData := &protocol.Series{
		Name:   &query.GetFromClause().Name,
		Fields: fields,
		Points: points,
	}
	yield(expectedData)
	return nil
}<commit_msg>get rid of executeCountQuery and assume that points belong to the same group if there was no group by clause.<commit_after>package engine

import (
	"common"
	"coordinator"
	"fmt"
	"os"
	"parser"
	"protocol"
	"runtime"
	"time"
)

const ALL_GROUP_IDENTIFIER = 1

type QueryEngine struct {
	coordinator coordinator.Coordinator
}

func (self *QueryEngine) RunQuery(query *parser.Query, yield func(*protocol.Series) error) (err error) {
	// don't let a panic pass beyond RunQuery
	defer func() {
		if err := recover(); err != nil {
			fmt.Fprintf(os.Stderr, "********************************BUG********************************\n")
			buf := make([]byte, 1024)
			n := runtime.Stack(buf, false)
			fmt.Fprintf(os.Stderr, "Error: %s. Stacktrace: %s\n", err, string(buf[:n]))
			err = common.NewQueryError(common.InternalError, "Internal Error")
		}
	}()

	if isCountQuery(query) {
		return self.executeCountQueryWithGroupBy(query, yield)
	} else {
		self.coordinator.DistributeQuery(query, yield)
	}
	return nil
}

func NewQueryEngine(c coordinator.Coordinator) (EngineI, error) {
	return &QueryEngine{c}, nil
}

func isCountQuery(query *parser.Query) bool {
	for _, column := range query.GetColumnNames() {
		if column.IsFunctionCall() && column.Name == "count" {
			return true
		}
	}

	return false
}

func getValueFromPoint(value *protocol.FieldValue, fType protocol.FieldDefinition_Type) interface{} {
	switch fType {
	case protocol.FieldDefinition_STRING:
		return *value.StringValue
	case protocol.FieldDefinition_INT32:
		return *value.IntValue
	case protocol.FieldDefinition_BOOL:
		return *value.BoolValue
	case protocol.FieldDefinition_DOUBLE:
		return *value.DoubleValue
	default:
		panic("WTF")
	}
}

func getTimestampFromPoint(window time.Duration, point *protocol.Point) int64 {
	return time.Unix(*point.Timestamp, 0).Round(window).Unix()
}

type Mapper func(*protocol.Point) interface{}
type InverseMapper func(interface{}, int) interface{}

// Returns a mapper and inverse mapper. A mapper is a function to map
// a point to a group and return an identifier of the group that can
// be used in a map (i.e. the returned interface must be hashable).
// An inverse mapper, takes a result of the mapper identifier and
// return the column values and/or timestamp bucket that defines the
// given group.
func createValuesToInterface(groupBy parser.GroupByClause, definitions []*protocol.FieldDefinition) (Mapper, InverseMapper, error) {
	// we shouldn't get an error, this is checked earlier in the executeCountQueryWithGroupBy
	window, _ := groupBy.GetGroupByTime()
	names := []string{}
	for _, value := range groupBy {
		if value.IsFunctionCall() {
			continue
		}
		names = append(names, value.Name)
	}

	switch len(names) {
	case 0:
		if window != nil {
			// this must be group by time
			return func(p *protocol.Point) interface{} {
					return getTimestampFromPoint(*window, p)
				}, func(i interface{}, idx int) interface{} {
					return i
				}, nil
		}

		// this must be group by time
		return func(p *protocol.Point) interface{} {
				return ALL_GROUP_IDENTIFIER
			}, func(i interface{}, idx int) interface{} {
				panic("This should never be called")
			}, nil

	case 1:
		// otherwise, find the type of the column and create a mapper
		idx := -1
		var fType protocol.FieldDefinition_Type
		for index, definition := range definitions {
			if *definition.Name == names[0] {
				idx = index
				fType = *definition.Type
				break
			}
		}

		if idx == -1 {
			return nil, nil, common.NewQueryError(common.InvalidArgument, fmt.Sprintf("Invalid column name %s", groupBy[0].Name))
		}

		if window != nil {
			return func(p *protocol.Point) interface{} {
					return [2]interface{}{
						getTimestampFromPoint(*window, p),
						getValueFromPoint(p.Values[idx], fType),
					}
				}, func(i interface{}, idx int) interface{} {
					return i.([2]interface{})[idx]
				}, nil
		}
		return func(p *protocol.Point) interface{} {
				return getValueFromPoint(p.Values[idx], fType)
			}, func(i interface{}, idx int) interface{} {
				return i
			}, nil

	case 2:
		names := []string{}
		for _, value := range groupBy {
			names = append(names, value.Name)
		}

		idx1, idx2 := -1, -1
		var fType1, fType2 protocol.FieldDefinition_Type

		for index, definition := range definitions {
			if *definition.Name == names[0] {
				idx1 = index
				fType1 = *definition.Type
			} else if *definition.Name == names[1] {
				idx2 = index
				fType2 = *definition.Type
			}
			if idx1 > 0 && idx2 > 0 {
				break
			}
		}

		if window != nil {
			return func(p *protocol.Point) interface{} {
					return [3]interface{}{
						getTimestampFromPoint(*window, p),
						getValueFromPoint(p.Values[idx1], fType1),
						getValueFromPoint(p.Values[idx2], fType2),
					}
				}, func(i interface{}, idx int) interface{} {
					return i.([3]interface{})[idx]
				}, nil
		}

		return func(p *protocol.Point) interface{} {
				return [2]interface{}{
					getValueFromPoint(p.Values[idx1], fType1),
					getValueFromPoint(p.Values[idx2], fType2),
				}
			}, func(i interface{}, idx int) interface{} {
				return i.([2]interface{})[idx]
			}, nil

	default:
		// TODO: return an error instead of killing the entire process
		return nil, nil, common.NewQueryError(common.InvalidArgument, "Group by currently support up to two columns and an optional group by time")
	}
}

func (self *QueryEngine) executeCountQueryWithGroupBy(query *parser.Query, yield func(*protocol.Series) error) error {
	counts := make(map[string]map[interface{}]int32)
	timestamps := make(map[string]map[interface{}]int64)

	groupBy := query.GetGroupByClause()

	fieldTypes := map[string]*protocol.FieldDefinition_Type{}
	var inverse InverseMapper

	duration, err := groupBy.GetGroupByTime()
	if err != nil {
		return err
	}

	err = self.coordinator.DistributeQuery(query, func(series *protocol.Series) error {
		var mapper Mapper
		mapper, inverse, err = createValuesToInterface(groupBy, series.Fields)
		if err != nil {
			return err
		}

		for _, field := range series.Fields {
			fieldTypes[*field.Name] = field.Type
		}

		for _, point := range series.Points {
			value := mapper(point)
			tableCounts := counts[*series.Name]
			if tableCounts == nil {
				tableCounts = make(map[interface{}]int32)
				counts[*series.Name] = tableCounts
			}
			tableCounts[value]++

			tableTimestamps := timestamps[*series.Name]
			if tableTimestamps == nil {
				tableTimestamps = make(map[interface{}]int64)
				timestamps[*series.Name] = tableTimestamps
			}
			if duration != nil {
				tableTimestamps[value] = getTimestampFromPoint(*duration, point)
			} else {
				tableTimestamps[value] = point.GetTimestamp()
			}
		}

		return nil
	})

	if err != nil {
		return err
	}

	expectedFieldType := protocol.FieldDefinition_INT32
	expectedName := "count"
	var sequenceNumber uint32 = 1

	/* fields := []*protocol.FieldDefinition{} */
	fields := []*protocol.FieldDefinition{
		&protocol.FieldDefinition{Name: &expectedName, Type: &expectedFieldType},
	}

	for _, value := range groupBy {
		if value.IsFunctionCall() {
			continue
		}

		tempName := value.Name
		fields = append(fields, &protocol.FieldDefinition{Name: &tempName, Type: fieldTypes[tempName]})
	}

	for table, tableCounts := range counts {
		tempTable := table
		points := []*protocol.Point{}
		for key, count := range tableCounts {
			tempKey := key
			tempCount := count

			timestamp := timestamps[table][tempKey]

			point := &protocol.Point{
				Timestamp:      &timestamp,
				SequenceNumber: &sequenceNumber,
				Values: []*protocol.FieldValue{
					&protocol.FieldValue{
						IntValue: &tempCount,
					},
				},
			}

			for idx, _ := range groupBy {
				value := inverse(tempKey, idx)

				switch x := value.(type) {
				case string:
					point.Values = append(point.Values, &protocol.FieldValue{StringValue: &x})
				case int32:
					point.Values = append(point.Values, &protocol.FieldValue{IntValue: &x})
				case bool:
					point.Values = append(point.Values, &protocol.FieldValue{BoolValue: &x})
				case float64:
					point.Values = append(point.Values, &protocol.FieldValue{DoubleValue: &x})
				}
			}

			points = append(points, point)
		}
		expectedData := &protocol.Series{
			Name:   &tempTable,
			Fields: fields,
			Points: points,
		}
		yield(expectedData)
	}

	return nil
}

func (self *QueryEngine) executeFunctionQueryWithGroupBy(query *parser.Query, yield func(*protocol.Series) error) error {
	results := make(map[interface{}]int32)
	timestamps := make(map[interface{}]int64)
	var functionType string
	var functionField string
	var inverse InverseMapper

	groupBy := query.GetGroupByClause()

	for _, column := range query.GetColumnNames() {
		if column.IsFunctionCall() && (column.Name == "min" || column.Name == "max") {
			functionType = column.Name
			functionField = column.Elems[0].Name
		}
	}

	fieldTypes := map[string]*protocol.FieldDefinition_Type{}

	duration, err := groupBy.GetGroupByTime()
	if err != nil {
		return err
	}

	self.coordinator.DistributeQuery(query, func(series *protocol.Series) error {
		var mapper Mapper
		mapper, inverse, err = createValuesToInterface(groupBy, series.Fields)
		var fieldIndex int

		for idx, field := range series.Fields {
			fieldTypes[*field.Name] = field.Type
			if *field.Name == functionField {
				fieldIndex = idx
			}
		}

		for _, point := range series.Points {
			key := mapper(point)

			if functionType == "min" {
				min := *point.Values[fieldIndex].IntValue

				if oldMin, exists := results[key]; exists {
					if min < oldMin {
						results[key] = min
					}
				} else {
					results[key] = min
				}
			}

			if functionType == "max" {
				max := *point.Values[fieldIndex].IntValue

				if oldMax, exists := results[key]; exists {
					if max > oldMax {
						results[key] = max
					}
				} else {
					results[key] = max
				}
			}

			if duration != nil {
				timestamps[key] = getTimestampFromPoint(*duration, point)
			} else {
				timestamps[key] = point.GetTimestamp()
			}
		}

		return nil
	})

	expectedFieldType := protocol.FieldDefinition_INT32
	expectedName := functionType
	var sequenceNumber uint32 = 1

	points := []*protocol.Point{}

	fields := []*protocol.FieldDefinition{
		&protocol.FieldDefinition{Name: &expectedName, Type: &expectedFieldType},
	}

	for _, value := range groupBy {
		if value.IsFunctionCall() {
			continue
		}

		tempName := value.Name
		fields = append(fields, &protocol.FieldDefinition{Name: &tempName, Type: fieldTypes[tempName]})
	}

	for key, result := range results {
		tempKey := key
		tempValue := result

		timestamp := timestamps[tempKey]

		point := &protocol.Point{
			Timestamp:      &timestamp,
			SequenceNumber: &sequenceNumber,
			Values: []*protocol.FieldValue{
				&protocol.FieldValue{
					IntValue: &tempValue,
				},
			},
		}

		for idx, _ := range groupBy {
			value := inverse(tempKey, idx)

			switch x := value.(type) {
			case string:
				point.Values = append(point.Values, &protocol.FieldValue{StringValue: &x})
			case int32:
				point.Values = append(point.Values, &protocol.FieldValue{IntValue: &x})
			case bool:
				point.Values = append(point.Values, &protocol.FieldValue{BoolValue: &x})
			case float64:
				point.Values = append(point.Values, &protocol.FieldValue{DoubleValue: &x})
			}
		}

		points = append(points, point)
	}

	expectedData := &protocol.Series{
		Name:   &query.GetFromClause().Name,
		Fields: fields,
		Points: points,
	}
	yield(expectedData)
	return nil
}
{'fraction_non_alphanumeric': 0.13306129317235288, 'fraction_numerical': 0.007067435110007993, 'mean_word_length': 3.717602698948204, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 12, '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': '6572102', 'n_tokens_mistral': 8753, 'n_tokens_neox': 7467, 'n_words': 1908}