<commit_before>package tsdb

import (
	"encoding/json"
	"fmt"
	"sort"

	"github.com/influxdb/influxdb/influxql"
)

// ShowMeasurementsExecutor implements the Executor interface for a SHOW MEASUREMENTS statement.
type ShowMeasurementsExecutor struct {
	stmt      *influxql.ShowMeasurementsStatement
	mappers   []Mapper
	chunkSize int
}

// NewShowMeasurementsExecutor returns a new ShowMeasurementsExecutor.
func NewShowMeasurementsExecutor(stmt *influxql.ShowMeasurementsStatement, mappers []Mapper, chunkSize int) *ShowMeasurementsExecutor {
	return &ShowMeasurementsExecutor{
		stmt:      stmt,
		mappers:   mappers,
		chunkSize: chunkSize,
	}
}

// Execute begins execution of the query and returns a channel to receive rows.
func (e *ShowMeasurementsExecutor) Execute() <-chan *influxql.Row {
	// Create output channel and stream data in a separate goroutine.
	out := make(chan *influxql.Row, 0)

	// It's important that all resources are released when execution completes.
	defer e.close()

	go func() {
		// Open the mappers.
		for _, m := range e.mappers {
			if err := m.Open(); err != nil {
				out <- &influxql.Row{Err: err}
				return
			}
		}

		// Create a set to hold measurement names from mappers.
		set := map[string]struct{}{}
		// Iterate through mappers collecting measurement names.
		for _, m := range e.mappers {
			// Get the data from the mapper.
			c, err := m.NextChunk()
			if err != nil {
				out <- &influxql.Row{Err: err}
				return
			} else if c == nil {
				// Mapper had no data.
				continue
			}

			// Convert the mapper chunk to a string array of measurement names.
			mms, ok := c.([]string)
			if !ok {
				out <- &influxql.Row{Err: fmt.Errorf("show measurements mapper returned invalid type: %T", c)}
				return
			}

			// Add the measurement names to the set.
			for _, mm := range mms {
				set[mm] = struct{}{}
			}
		}

		// Convert the set into an array of measurement names.
		measurements := make([]string, 0, len(set))
		for mm := range set {
			measurements = append(measurements, mm)
		}
		// Sort the names.
		sort.Strings(measurements)

		// Calculate OFFSET and LIMIT
		off := e.stmt.Offset
		lim := len(measurements)
		stmtLim := e.stmt.Limit

		if stmtLim > 0 && off+stmtLim < lim {
			lim = off + stmtLim
		} else if off > lim {
			off, lim = 0, 0
		}

		// Put the results in a row and send it.
		row := &influxql.Row{
			Name:    "measurements",
			Columns: []string{"name"},
			Values:  make([][]interface{}, 0, len(measurements)),
		}

		for _, m := range measurements[off:lim] {
			v := []interface{}{m}
			row.Values = append(row.Values, v)
		}

		if len(row.Values) > 0 {
			out <- row
		}

		close(out)
	}()
	return out
}

// Close closes the executor such that all resources are released. Once closed,
// an executor may not be re-used.
func (e *ShowMeasurementsExecutor) close() {
	if e != nil {
		for _, m := range e.mappers {
			m.Close()
		}
	}
}

// ShowMeasurementsMapper is a mapper for collecting measurement names from a shard.
type ShowMeasurementsMapper struct {
	remote    Mapper
	shard     *Shard
	stmt      *influxql.ShowMeasurementsStatement
	chunkSize int
	state     interface{}
}

// NewShowMeasurementsMapper returns a mapper for the given shard, which will return data for the meta statement.
func NewShowMeasurementsMapper(shard *Shard, stmt *influxql.ShowMeasurementsStatement, chunkSize int) *ShowMeasurementsMapper {
	return &ShowMeasurementsMapper{
		shard:     shard,
		stmt:      stmt,
		chunkSize: chunkSize,
	}
}

// Open opens the mapper for use.
func (m *ShowMeasurementsMapper) Open() error {
	if m.remote != nil {
		return m.remote.Open()
	}

	var measurements Measurements

	// If a WHERE clause was specified, filter the measurements.
	if m.stmt.Condition != nil {
		var err error
		measurements, err = m.shard.index.measurementsByExpr(m.stmt.Condition)
		if err != nil {
			return err
		}
	} else {
		// Otherwise, get all measurements from the database.
		measurements = m.shard.index.Measurements()
	}
	sort.Sort(measurements)

	// Create a channel to send measurement names on.
	ch := make(chan string)
	// Start a goroutine to send the names over the channel as needed.
	go func() {
		for _, mm := range measurements {
			ch <- mm.Name
		}<commit_msg>Fix panic with show measurements with partially replicated shards<commit_after>package tsdb

import (
	"encoding/json"
	"fmt"
	"sort"

	"github.com/influxdb/influxdb/influxql"
)

// ShowMeasurementsExecutor implements the Executor interface for a SHOW MEASUREMENTS statement.
type ShowMeasurementsExecutor struct {
	stmt      *influxql.ShowMeasurementsStatement
	mappers   []Mapper
	chunkSize int
}

// NewShowMeasurementsExecutor returns a new ShowMeasurementsExecutor.
func NewShowMeasurementsExecutor(stmt *influxql.ShowMeasurementsStatement, mappers []Mapper, chunkSize int) *ShowMeasurementsExecutor {
	return &ShowMeasurementsExecutor{
		stmt:      stmt,
		mappers:   mappers,
		chunkSize: chunkSize,
	}
}

// Execute begins execution of the query and returns a channel to receive rows.
func (e *ShowMeasurementsExecutor) Execute() <-chan *influxql.Row {
	// Create output channel and stream data in a separate goroutine.
	out := make(chan *influxql.Row, 0)

	// It's important that all resources are released when execution completes.
	defer e.close()

	go func() {
		// Open the mappers.
		for _, m := range e.mappers {
			if err := m.Open(); err != nil {
				out <- &influxql.Row{Err: err}
				return
			}
		}

		// Create a set to hold measurement names from mappers.
		set := map[string]struct{}{}
		// Iterate through mappers collecting measurement names.
		for _, m := range e.mappers {
			// Get the data from the mapper.
			c, err := m.NextChunk()
			if err != nil {
				out <- &influxql.Row{Err: err}
				return
			} else if c == nil {
				// Mapper had no data.
				continue
			}

			// Convert the mapper chunk to a string array of measurement names.
			mms, ok := c.([]string)
			if !ok {
				out <- &influxql.Row{Err: fmt.Errorf("show measurements mapper returned invalid type: %T", c)}
				return
			}

			// Add the measurement names to the set.
			for _, mm := range mms {
				set[mm] = struct{}{}
			}
		}

		// Convert the set into an array of measurement names.
		measurements := make([]string, 0, len(set))
		for mm := range set {
			measurements = append(measurements, mm)
		}
		// Sort the names.
		sort.Strings(measurements)

		// Calculate OFFSET and LIMIT
		off := e.stmt.Offset
		lim := len(measurements)
		stmtLim := e.stmt.Limit

		if stmtLim > 0 && off+stmtLim < lim {
			lim = off + stmtLim
		} else if off > lim {
			off, lim = 0, 0
		}

		// Put the results in a row and send it.
		row := &influxql.Row{
			Name:    "measurements",
			Columns: []string{"name"},
			Values:  make([][]interface{}, 0, len(measurements)),
		}

		for _, m := range measurements[off:lim] {
			v := []interface{}{m}
			row.Values = append(row.Values, v)
		}

		if len(row.Values) > 0 {
			out <- row
		}

		close(out)
	}()
	return out
}

// Close closes the executor such that all resources are released. Once closed,
// an executor may not be re-used.
func (e *ShowMeasurementsExecutor) close() {
	if e != nil {
		for _, m := range e.mappers {
			m.Close()
		}
	}
}

// ShowMeasurementsMapper is a mapper for collecting measurement names from a shard.
type ShowMeasurementsMapper struct {
	remote    Mapper
	shard     *Shard
	stmt      *influxql.ShowMeasurementsStatement
	chunkSize int
	state     interface{}
}

// NewShowMeasurementsMapper returns a mapper for the given shard, which will return data for the meta statement.
func NewShowMeasurementsMapper(shard *Shard, stmt *influxql.ShowMeasurementsStatement, chunkSize int) *ShowMeasurementsMapper {
	return &ShowMeasurementsMapper{
		shard:     shard,
		stmt:      stmt,
		chunkSize: chunkSize,
	}
}

// Open opens the mapper for use.
func (m *ShowMeasurementsMapper) Open() error {
	if m.remote != nil {
		return m.remote.Open()
	}

	var measurements Measurements

	if m.shard != nil {
		// If a WHERE clause was specified, filter the measurements.
		if m.stmt.Condition != nil {
			var err error
			measurements, err = m.shard.index.measurementsByExpr(m.stmt.Condition)
			if err != nil {
				return err
			}
		} else {
			// Otherwise, get all measurements from the database.
			measurements = m.shard.index.Measurements()
		}
		sort.Sort(measurements)
	}

	// Create a channel to send measurement names on.
	ch := make(chan string)
	// Start a goroutine to send the names over the channel as needed.
	go func() {
		for _, mm := range measurements {
			ch <- mm.Name
		}
{'fraction_non_alphanumeric': 0.10867038586703859, 'fraction_numerical': 0.0016271501627150163, 'mean_word_length': 3.144990366088632, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 9, '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': '485391', 'n_tokens_mistral': 3046, 'n_tokens_neox': 2713, 'n_words': 915}