<commit_before>package main_test

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"regexp"
	"runtime"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/influxdb/influxdb/client"
	main "github.com/influxdb/influxdb/cmd/influxd"
)

func init() {
	fmt.Printf("System Info: CPU=%d GOMAXPROCS=%d\n", runtime.NumCPU(), runtime.GOMAXPROCS(0))
}

const (
	// Use a prime batch size, so that internal batching code, which most likely
	// uses nice round batches, has to deal with leftover.
	batchSize = 4217

	openTSDBTestTimeout = 5 * time.Second
	graphiteTestTimeout = 5 * time.Second
)

type writeFn func(t *testing.T, node *TestNode, database, retention string)

// tempfile returns a temporary path.
func tempfile() string {
	f, _ := ioutil.TempFile("", "influxdb-")
	path := f.Name()
	f.Close()
	os.Remove(path)
	return path
}

// urlFor returns a URL with the path and query params correctly appended and set.
func urlFor(u *url.URL, path string, params url.Values) *url.URL {
	v, _ := url.Parse(u.String())
	v.Path = path
	v.RawQuery = params.Encode()
	return v
}

// rewriteDbRp returns a copy of old with occurrences of %DB% with the given database,
// and occurences of %RP with the given retention
func rewriteDbRp(old, database, retention string) string {
	return strings.Replace(strings.Replace(old, "%DB%", database, -1), "%RP%", retention, -1)
}

// Node represents a node under test, which is both a broker and data node.
type TestNode struct {
	node   *main.Node
	url    *url.URL
	leader bool
}

// Cluster represents a multi-node cluster.
type Cluster []*TestNode

func (c *Cluster) Close() {
	for _, n := range *c {
		if err := n.node.Close(); err != nil {
			log.Fatalf("failed to close node: %s", err)
		}
	}
}

// createCombinedNodeCluster creates a cluster of nServers nodes, each of which
// runs as both a Broker and Data node. If any part cluster creation fails,
// the testing is marked as failed.
//
// This function returns a slice of nodes, the first of which will be the leader.
func createCombinedNodeCluster(t *testing.T, testName, tmpDir string, nNodes int, baseConfig *main.Config) Cluster {
	t.Logf("Creating cluster of %d nodes for test %s", nNodes, testName)
	if nNodes < 1 {
		t.Fatalf("Test %s: asked to create nonsense cluster", testName)
	}

	nodes := make([]*TestNode, 0)

	tmpBrokerDir := filepath.Join(tmpDir, "broker-integration-test")
	tmpDataDir := filepath.Join(tmpDir, "data-integration-test")
	t.Logf("Test %s: using tmp directory %q for brokers\n", testName, tmpBrokerDir)
	t.Logf("Test %s: using tmp directory %q for data nodes\n", testName, tmpDataDir)
	// Sometimes if a test fails, it's because of a log.Fatal() in the program.
	// This prevents the defer from cleaning up directories.
	// To be safe, nuke them always before starting
	_ = os.RemoveAll(tmpBrokerDir)
	_ = os.RemoveAll(tmpDataDir)

	// Create the first node, special case.
	c := baseConfig
	if c == nil {
		c, _ = main.NewTestConfig()
	}
	basePort := 0
	c.Broker.Dir = filepath.Join(tmpBrokerDir, strconv.Itoa(basePort))
	c.Data.Dir = filepath.Join(tmpDataDir, strconv.Itoa(basePort))
	c.Port = basePort
	c.Admin.Port = 0
	c.Admin.Enabled = false
	c.ReportingDisabled = true
	c.Snapshot.Enabled = false
	c.Logging.HTTPAccess = false

	cmd := main.NewRunCommand()
	baseNode := cmd.Open(c, "")
	b := baseNode.Broker
	s := baseNode.DataNode

	if b == nil {
		t.Fatalf("Test %s: failed to create broker on port %d", testName, basePort)
	}
	if s == nil {
		t.Fatalf("Test %s: failed to create leader data node on port %d", testName, basePort)
	}
	nodes = append(nodes, &TestNode{
		node:   baseNode,
		url:    baseNode.ClusterURL(),
		leader: true,
	})
	t.Log(baseNode.ClusterURL())

	// Create subsequent nodes, which join to first node.
	for i := 1; i < nNodes; i++ {
		c.Broker.Dir = filepath.Join(tmpBrokerDir, strconv.Itoa(i))
		c.Data.Dir = filepath.Join(tmpDataDir, strconv.Itoa(i))
		c.Port = 0

		cmd := main.NewRunCommand()
		node := cmd.Open(c, baseNode.ClusterURL().String())
		if node.Broker == nil {
			t.Fatalf("Test %s: failed to create following broker on addr %s", testName, node.ClusterURL().String())
		}
		if node.DataNode == nil {
			t.Fatalf("Test %s: failed to create following data node on addr %s", testName, node.ClusterURL().String())
		}

		nodes = append(nodes, &TestNode{
			node: node,
			url:  node.ClusterURL(),
		})
		t.Log(node.ClusterURL())

	}

	return nodes
}

// createDatabase creates a database, and verifies that the creation was successful.
func createDatabase(t *testing.T, testName string, nodes Cluster, database string) {
	t.Logf("Test: %s: creating database %s", testName, database)
	query(t, nodes[:1], "", "CREATE DATABASE "+database, `{"results":[{}]}`, "")
}

// createRetentionPolicy creates a retetention policy and verifies that the creation was successful.
// Replication factor is set to equal the number nodes in the cluster.
func createRetentionPolicy(t *testing.T, testName string, nodes Cluster, database, retention string, replicationFactor int) {
	t.Logf("Creating retention policy %s for database %s", retention, database)
	command := fmt.Sprintf("CREATE RETENTION POLICY %s ON %s DURATION 1h REPLICATION %d DEFAULT", retention, database, replicationFactor)
	query(t, nodes[:1], "", command, `{"results":[{}]}`, "")
}

// deleteDatabase delete a database, and verifies that the deletion was successful.
func deleteDatabase(t *testing.T, testName string, nodes Cluster, database string) {
	t.Logf("Test: %s: deleting database %s", testName, database)
	query(t, nodes[:1], "", "DROP DATABASE "+database, `{"results":[{}]}`, "")
}

// dumpClusterDiags dumps the diagnostics of each node.
func dumpClusterDiags(t *testing.T, testName string, nodes Cluster) {
	t.Logf("Test: %s: dumping node diagnostics", testName)
	for _, n := range nodes {
		r, _, _ := query(t, Cluster{n}, "", "SHOW DIAGNOSTICS", "", "")
		t.Log(r)
	}
}

// dumpClusterStats dumps the statistics of each node.
func dumpClusterStats(t *testing.T, testName string, nodes Cluster) {
	t.Logf("Test: %s: dumping node stats", testName)
	for _, n := range nodes {
		r, _, _ := query(t, Cluster{n}, "", "SHOW STATS", "", "")
		t.Log(r)
	}
}

// writes writes the provided data to the cluster. It verfies that a 200 OK is returned by the server.
func write(t *testing.T, node *TestNode, data string) {
	u := urlFor(node.url, "write", url.Values{})

	resp, err := http.Post(u.String(), "application/json", bytes.NewReader([]byte(data)))
	if err != nil {
		t.Fatalf("Couldn't write data: %s", err)
	}
	defer resp.Body.Close()
	body, _ := ioutil.ReadAll(resp.Body)
	if string(body) != "" {
		t.Log("BODY: ", string(body))
	}
	if resp.StatusCode != http.StatusOK {
		body, _ := ioutil.ReadAll(resp.Body)
		t.Fatalf("Write to database failed.  Unexpected status code.  expected: %d, actual %d, %s", http.StatusOK, resp.StatusCode, string(body))
	}
}

// query executes the given query against all nodes in the cluster, and verifies no errors occured, and
// ensures the returned data is as expected
func query(t *testing.T, nodes Cluster, urlDb, query, expected, expectPattern string) (string, bool, int) {
	v := url.Values{"q": []string{query}}
	if urlDb != "" {
		v.Set("db", urlDb)
	}

	var actual string
	// Query the data exists
	for i, n := range nodes {
		u := urlFor(n.url, "query", v)
		resp, err := http.Get(u.String())
		if err != nil {
			t.Fatalf("Failed to execute query '%s': %s", query, err.Error())
		}
		defer resp.Body.Close()

		body, err := ioutil.ReadAll(resp.Body)
		if err != nil {
			t.Fatalf("Couldn't read body of response: %s", err.Error())
		}
		actual = string(body)

		if expected != "" && expected != actual {
			return actual, false, i
		} else if expectPattern != "" {
			re := regexp.MustCompile(expectPattern)
			if !re.MatchString(actual) {
				return actual, false, i
			}
		}
	}

	return actual, true, len(nodes)
}

// queryAndWait executes the given query against all nodes in the cluster, and verifies no errors occured, and
// ensures the returned data is as expected until the timeout occurs
func queryAndWait(t *testing.T, nodes Cluster, urlDb, q, expected, expectPattern string, timeout time.Duration) (string, bool, int) {
	v := url.Values{"q": []string{q}}
	if urlDb != "" {
		v.Set("db", urlDb)
	}

	sleep := 100 * time.Millisecond
	// Check to see if they set the env for duration sleep
	if sleepRaw := os.Getenv("TEST_SLEEP"); sleepRaw != "" {
		d, err := time.ParseDuration(sleepRaw)
		if err != nil {
			t.Fatal("Invalid TEST_SLEEP value:", err)
		}
		// this will limit the http log noise in the test output
		sleep = d
		timeout = d + 1
	}

	var done <-chan time.Time
	if timeout > 0 {
		timer := time.NewTimer(timeout)
		done = timer.C
		defer timer.Stop()
	}

	nQueriedOK := 0
	for {
		got, ok, n := query(t, nodes, urlDb, q, expected, expectPattern)
		if n > nQueriedOK {
			nQueriedOK = n
		}
		if ok {
			return got, ok, nQueriedOK
		}
		select {
		case <-done:
			return got, false, nQueriedOK
		case <-time.After(sleep):
		}
	}
}

// mergeMany ensures that when merging many series together and some of them have a different number
// of points than others in a group by interval the results are correct
var mergeMany = func(t *testing.T, node *TestNode, database, retention string) {
	for i := 1; i < 11; i++ {
		for j := 1; j < 5+i%3; j++ {
			data := fmt.Sprintf(`{"database": "%s", "retentionPolicy": "%s", "points": [{"name": "cpu", "timestamp": "%s", "tags": {"host": "server_%d"}, "fields": {"value": 22}}]}`,
				database, retention, time.Unix(int64(j), int64(0)).UTC().Format(time.RFC3339), i)
			write(t, node, data)
		}
	}
}

var limitAndOffset = func(t *testing.T, node *TestNode, database, retention string) {
	for i := 1; i < 10; i++ {
		data := fmt.Sprintf(`{"database": "%s", "retentionPolicy": "%s", "points": [{"name": "cpu", "timestamp": "%s", "tags": {"region": "us-east", "host": "server-%d"}, "fields": {"value": %d}}]}`,
			database, retention, time.Unix(int64(i), int64(0)).Format(time.RFC3339), i, i)
		write(t, node, data)
	}
}

func runTest_rawDataReturnsInOrder(t *testing.T, testName string, nodes Cluster, database, retention string, replicationFactor int) {
	// skip this test if they're just looking to run some of thd data tests
	if os.Getenv("TEST_PREFIX") != "" {
		return
	}
	t.Logf("Running %s:rawDataReturnsInOrder against %d-node cluster", testName, len(nodes))

	// Start by ensuring database and retention policy exist.
	createDatabase(t, testName, nodes, database)
	createRetentionPolicy(t, testName, nodes, database, retention, replicationFactor)
	numPoints := 500
	var expected string

	for i := 1; i < numPoints; i++ {
		data := fmt.Sprintf(`{"database": "%s", "retentionPolicy": "%s", "points": [{"name": "cpu", "timestamp": "%s", "tags": {"region": "us-east", "host": "server-%d"}, "fields": {"value": %d}}]}`,
			database, retention, time.Unix(int64(i), int64(0)).Format(time.RFC3339), i%10, i)
		write(t, nodes[0], data)
	}

	expected = fmt.Sprintf(`{"results":[{"series":[{"name":"cpu","columns":["time","count"],"values":[["1970-01-01T00:00:00Z",%d]]}]}]}`, numPoints-1)
	got, ok, nOK := queryAndWait(t, nodes, database, `SELECT count(value) FROM cpu`, expected, "", 120*time.Second)
	if !ok {
		t.Errorf("test %s:rawDataReturnsInOrder failed, SELECT count() query returned unexpected data\nexp: %s\n, got: %s\n%d nodes responded correctly", testName, expected, got, nOK)
		dumpClusterDiags(t, testName, nodes)
		dumpClusterStats(t, testName, nodes)
	}

	// Create expected JSON string dynamically.
	expectedValues := make([]string, 0)
	for i := 1; i < numPoints; i++ {
		expectedValues = append(expectedValues, fmt.Sprintf(`["%s",%d]`, time.Unix(int64(i), int64(0)).UTC().Format(time.RFC3339), i))
	}
	expected = fmt.Sprintf(`{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[%s]}]}]}`, strings.Join(expectedValues, ","))
	got, ok, nOK = query(t, nodes, database, `SELECT value FROM cpu`, expected, "")
	if !ok {
		t.Errorf("test %s failed, SELECT query returned unexpected data\nexp: %s\ngot: %s\n%d nodes responded correctly", testName, expected, got, nOK)
		dumpClusterDiags(t, testName, nodes)
		dumpClusterStats(t, testName, nodes)
	}
}

// runTests_Errors tests some basic error cases.
func runTests_Errors(t *testing.T, nodes Cluster) {
	t.Logf("Running tests against %d-node cluster", len(nodes))

	tests := []struct {
		name     string
		write    string // If equal to the empty string, no data is written.
		query    string // If equal to the blank string, no query is executed.
		expected string // If 'query' is equal to the blank string, this is ignored.
	}{
		{
			name:     "simple SELECT from non-existent database",
			write:    "",
			query:    `SELECT * FROM "qux"."bar".cpu`,
			expected: `{"results":[{"error":"database not found: qux"}]}`,
		},
	}

	for _, tt := range tests {
		if tt.write != "" {
			write(t, nodes[0], tt.write)
		}

		if tt.query != "" {
			got, ok, nOK := query(t, nodes, "", tt.query, tt.expected, "")
			if !ok {
				t.Errorf("Test '%s' failed, expected: %s, got: %s, %d nodes responded correctly", tt.name, tt.expected, got, nOK)
			}
		}
	}
}

// runTests tests write and query of data. Setting testNumbers allows only a subset of tests to be run.
func runTestsData(t *testing.T, testName string, nodes Cluster, database, retention string, replicationFactor int) {
	t.Logf("Running tests against %d-node cluster", len(nodes))

	yesterday := time.Now().Add(-1 * time.Hour * 24).UTC()
	now := time.Now().UTC()

	// Start by ensuring database and retention policy exist.
	createDatabase(t, testName, nodes, database)
	createRetentionPolicy(t, testName, nodes, database, retention, replicationFactor)

	// The tests. Within these tests %DB% and %RP% will be replaced with the database and retention passed into
	// this function.
	tests := []struct {
		skip          bool    // Skip the test.
		reset         bool    // Delete and recreate the database.
		name          string  // Test name, for easy-to-read test log output.
		write         string  // If equal to the empty string, no data is written.
		writeFn       writeFn // If non-nil, called after 'write' data (if any) is written.
		query         string  // If equal to the blank string, no query is executed.
		queryDb       string  // If set, is used as the "db" query param.
		queryOne      bool    // If set, only 1 node is queried.
		expected      string  // If 'query' is equal to the blank string, this is ignored.
		expectPattern string  // Regexp alternative to expected field. (ignored if expected is set)
	}{
		// Data read and write tests
		{
			reset:    true,
			name:     "single point with timestamp",
			write:    `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "cpu", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": 100}}]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".cpu`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",100]]}]}]}`,
		},
		{
			name:     "single point count query with timestamp",
			query:    `SELECT count(value) FROM "%DB%"."%RP%".cpu`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]}]}]}`,
		},
		{
			name:     "single string point with timestamp",
			write:    `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "logs", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": "disk full"}}]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".logs`,
			expected: `{"results":[{"series":[{"name":"logs","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z","disk full"]]}]}]}`,
		},
		{
			name:     "single bool point with timestamp",
			write:    `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "status", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": "true"}}]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".status`,
			expected: `{"results":[{"series":[{"name":"status","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z","true"]]}]}]}`,
		},
		{
			name:     "single point, select with now()",
			query:    `SELECT * FROM "%DB%"."%RP%".cpu WHERE time < now()`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",100]]}]}]}`,
		},
		{
			name:     "single point, select with now(), two queries",
			query:    `SELECT * FROM "%DB%"."%RP%".cpu WHERE time < now();SELECT * FROM "%DB%"."%RP%".cpu WHERE time < now()`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",100]]}]},{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",100]]}]}]}`,
		},

		{
			name:          "measurement not found",
			query:         `SELECT value FROM "%DB%"."%RP%".foobarbaz`,
			expectPattern: `.*measurement not found.*`,
		},
		{
			name:     "field not found",
			query:    `SELECT abc FROM "%DB%"."%RP%".cpu WHERE time < now()`,
			expected: `{"results":[{"error":"unknown field or tag name in select clause: abc"}]}`,
		},
		{
			name:     "empty result",
			query:    `SELECT value FROM cpu WHERE time >= '3000-01-01 00:00:05'`,
			queryDb:  "%DB%",
			expected: `{"results":[{}]}`,
		},

		{
			reset: true,
			name:  "two points with timestamp",
			write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "cpu", "timestamp": "2015-02-28T01:04:36.703820946Z", "fields": {"value": 100}},
		                                                                         {"name": "cpu", "timestamp": "2015-02-28T01:03:36.703820946Z", "fields": {"value": 100}}
		                                                                        ]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".cpu`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",100],["2015-02-28T01:04:36.703820946Z",100]]}]}]}`,
		},

		{
			reset: true,
			name:  "two points with negative values",
			write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "cpu", "timestamp": "2015-02-28T01:04:36.703820946Z", "fields": {"value": -200}},
                                                                                 {"name": "cpu", "timestamp": "2015-02-28T01:03:36.703820946Z", "fields": {"value": -100}}
                                                                                ]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".cpu`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",-100],["2015-02-28T01:04:36.703820946Z",-200]]}]}]}`,
		},

		// Data read and write tests using relative time
		{
			reset:    true,
			name:     "single point with timestamp pre-calculated for past time queries yesterday",
			write:    `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "cpu", "timestamp": "` + yesterday.Format(time.RFC3339Nano) + `", "tags": {"host": "server01"}, "fields": {"value": 100}}]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".cpu where time >= '` + yesterday.Add(-1*time.Minute).Format(time.RFC3339Nano) + `'`,
			expected: fmt.Sprintf(`{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["%s",100]]}]}]}`, yesterday.Format(time.RFC3339Nano)),
		},
		{
			reset:    true,
			name:     "single point with timestamp pre-calculated for relative time queries now",
			write:    `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "cpu", "timestamp": "` + now.Format(time.RFC3339Nano) + `", "tags": {"host": "server01"}, "fields": {"value": 100}}]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".cpu where time >= now() - 1m`,
			expected: fmt.Sprintf(`{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["%s",100]]}]}]}`, now.Format(time.RFC3339Nano)),
		},

		// Merge tests.
		{
			reset:    true,
			name:     "merge many",
			writeFn:  mergeMany,
			query:    `SELECT count(value) FROM cpu WHERE time >= '1970-01-01T00:00:01Z' AND time <= '1970-01-01T00:00:06Z' GROUP BY time(1s)`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","count"],"values":[["1970-01-01T00:00:01Z",10],["1970-01-01T00:00:02Z",10],["1970-01-01T00:00:03Z",10],["1970-01-01T00:00:04Z",10],["1970-01-01T00:00:05Z",7],["1970-01-01T00:00:06Z",3]]}]}]}`,
		},

		// Limit and offset
		{
			reset:    true,
			name:     "limit and offset",
			writeFn:  limitAndOffset,
			query:    `SELECT count(value) FROM cpu GROUP BY * SLIMIT 2 SOFFSET 1`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu","tags":{"host":"server-2","region":"us-east"},"columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]},{"name":"cpu","tags":{"host":"server-3","region":"us-east"},"columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]}]}]}`,
		},
		{
			query:    `SELECT count(value) FROM cpu GROUP BY * SLIMIT 2 SOFFSET 3`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu","tags":{"host":"server-4","region":"us-east"},"columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]},{"name":"cpu","tags":{"host":"server-5","region":"us-east"},"columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]}]}]}`,
		},
		{
			query:    `SELECT count(value) FROM cpu GROUP BY * SLIMIT 3 SOFFSET 8`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu","tags":{"host":"server-9","region":"us-east"},"columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]}]}]}`,
		},

		// FROM regex tests
		{
			reset: true,
			name:  "FROM regex using default db and rp",
			write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [
				{"name": "cpu1", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": 10}},
				{"name": "cpu2", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": 20}},
				{"name": "cpu3", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": 30}}
			]}`,
			query:    `SELECT * FROM /cpu[13]/`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu1","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",10]]},{"name":"cpu3","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",30]]}]}]}`,
		},
		{
			name:     `FROM regex specifying db and rp`,
			query:    `SELECT * FROM "%DB%"."%RP%"./cpu[13]/`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu1","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",10]]},{"name":"cpu3","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",30]]}]}]}`,
		},
		{
			name:     `FROM regex using specified db and default rp`,
			query:    `SELECT * FROM "%DB%"../cpu[13]/`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu1","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",10]]},{"name":"cpu3","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",30]]}]}]}`,
		},
		{
			name:     `FROM regex using default db and specified rp`,
			query:    `SELECT * FROM "%RP%"./cpu[13]/`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu1","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",10]]},{"name":"cpu3","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",30]]}]}]}`,
		},

		// Aggregations
		{
			reset: true,
			name:  "aggregations",
			write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [
				{"name": "cpu", "timestamp": "2000-01-01T00:00:00Z", "tags": {"region": "us-east"}, "fields": {"value": 20}},<commit_msg>add test for large mean aggregations<commit_after>package main_test

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"math"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"regexp"
	"runtime"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/influxdb/influxdb/client"
	main "github.com/influxdb/influxdb/cmd/influxd"
)

func init() {
	fmt.Printf("System Info: CPU=%d GOMAXPROCS=%d\n", runtime.NumCPU(), runtime.GOMAXPROCS(0))
}

const (
	// Use a prime batch size, so that internal batching code, which most likely
	// uses nice round batches, has to deal with leftover.
	batchSize = 4217

	openTSDBTestTimeout = 5 * time.Second
	graphiteTestTimeout = 5 * time.Second
)

type writeFn func(t *testing.T, node *TestNode, database, retention string)

// tempfile returns a temporary path.
func tempfile() string {
	f, _ := ioutil.TempFile("", "influxdb-")
	path := f.Name()
	f.Close()
	os.Remove(path)
	return path
}

// urlFor returns a URL with the path and query params correctly appended and set.
func urlFor(u *url.URL, path string, params url.Values) *url.URL {
	v, _ := url.Parse(u.String())
	v.Path = path
	v.RawQuery = params.Encode()
	return v
}

// rewriteDbRp returns a copy of old with occurrences of %DB% with the given database,
// and occurences of %RP with the given retention
func rewriteDbRp(old, database, retention string) string {
	return strings.Replace(strings.Replace(old, "%DB%", database, -1), "%RP%", retention, -1)
}

// Node represents a node under test, which is both a broker and data node.
type TestNode struct {
	node   *main.Node
	url    *url.URL
	leader bool
}

// Cluster represents a multi-node cluster.
type Cluster []*TestNode

func (c *Cluster) Close() {
	for _, n := range *c {
		if err := n.node.Close(); err != nil {
			log.Fatalf("failed to close node: %s", err)
		}
	}
}

// createCombinedNodeCluster creates a cluster of nServers nodes, each of which
// runs as both a Broker and Data node. If any part cluster creation fails,
// the testing is marked as failed.
//
// This function returns a slice of nodes, the first of which will be the leader.
func createCombinedNodeCluster(t *testing.T, testName, tmpDir string, nNodes int, baseConfig *main.Config) Cluster {
	t.Logf("Creating cluster of %d nodes for test %s", nNodes, testName)
	if nNodes < 1 {
		t.Fatalf("Test %s: asked to create nonsense cluster", testName)
	}

	nodes := make([]*TestNode, 0)

	tmpBrokerDir := filepath.Join(tmpDir, "broker-integration-test")
	tmpDataDir := filepath.Join(tmpDir, "data-integration-test")
	t.Logf("Test %s: using tmp directory %q for brokers\n", testName, tmpBrokerDir)
	t.Logf("Test %s: using tmp directory %q for data nodes\n", testName, tmpDataDir)
	// Sometimes if a test fails, it's because of a log.Fatal() in the program.
	// This prevents the defer from cleaning up directories.
	// To be safe, nuke them always before starting
	_ = os.RemoveAll(tmpBrokerDir)
	_ = os.RemoveAll(tmpDataDir)

	// Create the first node, special case.
	c := baseConfig
	if c == nil {
		c, _ = main.NewTestConfig()
	}
	basePort := 0
	c.Broker.Dir = filepath.Join(tmpBrokerDir, strconv.Itoa(basePort))
	c.Data.Dir = filepath.Join(tmpDataDir, strconv.Itoa(basePort))
	c.Port = basePort
	c.Admin.Port = 0
	c.Admin.Enabled = false
	c.ReportingDisabled = true
	c.Snapshot.Enabled = false
	c.Logging.HTTPAccess = false

	cmd := main.NewRunCommand()
	baseNode := cmd.Open(c, "")
	b := baseNode.Broker
	s := baseNode.DataNode

	if b == nil {
		t.Fatalf("Test %s: failed to create broker on port %d", testName, basePort)
	}
	if s == nil {
		t.Fatalf("Test %s: failed to create leader data node on port %d", testName, basePort)
	}
	nodes = append(nodes, &TestNode{
		node:   baseNode,
		url:    baseNode.ClusterURL(),
		leader: true,
	})
	t.Log(baseNode.ClusterURL())

	// Create subsequent nodes, which join to first node.
	for i := 1; i < nNodes; i++ {
		c.Broker.Dir = filepath.Join(tmpBrokerDir, strconv.Itoa(i))
		c.Data.Dir = filepath.Join(tmpDataDir, strconv.Itoa(i))
		c.Port = 0

		cmd := main.NewRunCommand()
		node := cmd.Open(c, baseNode.ClusterURL().String())
		if node.Broker == nil {
			t.Fatalf("Test %s: failed to create following broker on addr %s", testName, node.ClusterURL().String())
		}
		if node.DataNode == nil {
			t.Fatalf("Test %s: failed to create following data node on addr %s", testName, node.ClusterURL().String())
		}

		nodes = append(nodes, &TestNode{
			node: node,
			url:  node.ClusterURL(),
		})
		t.Log(node.ClusterURL())

	}

	return nodes
}

// createDatabase creates a database, and verifies that the creation was successful.
func createDatabase(t *testing.T, testName string, nodes Cluster, database string) {
	t.Logf("Test: %s: creating database %s", testName, database)
	query(t, nodes[:1], "", "CREATE DATABASE "+database, `{"results":[{}]}`, "")
}

// createRetentionPolicy creates a retetention policy and verifies that the creation was successful.
// Replication factor is set to equal the number nodes in the cluster.
func createRetentionPolicy(t *testing.T, testName string, nodes Cluster, database, retention string, replicationFactor int) {
	t.Logf("Creating retention policy %s for database %s", retention, database)
	command := fmt.Sprintf("CREATE RETENTION POLICY %s ON %s DURATION 1h REPLICATION %d DEFAULT", retention, database, replicationFactor)
	query(t, nodes[:1], "", command, `{"results":[{}]}`, "")
}

// deleteDatabase delete a database, and verifies that the deletion was successful.
func deleteDatabase(t *testing.T, testName string, nodes Cluster, database string) {
	t.Logf("Test: %s: deleting database %s", testName, database)
	query(t, nodes[:1], "", "DROP DATABASE "+database, `{"results":[{}]}`, "")
}

// dumpClusterDiags dumps the diagnostics of each node.
func dumpClusterDiags(t *testing.T, testName string, nodes Cluster) {
	t.Logf("Test: %s: dumping node diagnostics", testName)
	for _, n := range nodes {
		r, _, _ := query(t, Cluster{n}, "", "SHOW DIAGNOSTICS", "", "")
		t.Log(r)
	}
}

// dumpClusterStats dumps the statistics of each node.
func dumpClusterStats(t *testing.T, testName string, nodes Cluster) {
	t.Logf("Test: %s: dumping node stats", testName)
	for _, n := range nodes {
		r, _, _ := query(t, Cluster{n}, "", "SHOW STATS", "", "")
		t.Log(r)
	}
}

// writes writes the provided data to the cluster. It verfies that a 200 OK is returned by the server.
func write(t *testing.T, node *TestNode, data string) {
	u := urlFor(node.url, "write", url.Values{})

	resp, err := http.Post(u.String(), "application/json", bytes.NewReader([]byte(data)))
	if err != nil {
		t.Fatalf("Couldn't write data: %s", err)
	}
	defer resp.Body.Close()
	body, _ := ioutil.ReadAll(resp.Body)
	if string(body) != "" {
		t.Log("BODY: ", string(body))
	}
	if resp.StatusCode != http.StatusOK {
		body, _ := ioutil.ReadAll(resp.Body)
		t.Fatalf("Write to database failed.  Unexpected status code.  expected: %d, actual %d, %s", http.StatusOK, resp.StatusCode, string(body))
	}
}

// query executes the given query against all nodes in the cluster, and verifies no errors occured, and
// ensures the returned data is as expected
func query(t *testing.T, nodes Cluster, urlDb, query, expected, expectPattern string) (string, bool, int) {
	v := url.Values{"q": []string{query}}
	if urlDb != "" {
		v.Set("db", urlDb)
	}

	var actual string
	// Query the data exists
	for i, n := range nodes {
		u := urlFor(n.url, "query", v)
		resp, err := http.Get(u.String())
		if err != nil {
			t.Fatalf("Failed to execute query '%s': %s", query, err.Error())
		}
		defer resp.Body.Close()

		body, err := ioutil.ReadAll(resp.Body)
		if err != nil {
			t.Fatalf("Couldn't read body of response: %s", err.Error())
		}
		actual = string(body)

		if expected != "" && expected != actual {
			return actual, false, i
		} else if expectPattern != "" {
			re := regexp.MustCompile(expectPattern)
			if !re.MatchString(actual) {
				return actual, false, i
			}
		}
	}

	return actual, true, len(nodes)
}

// queryAndWait executes the given query against all nodes in the cluster, and verifies no errors occured, and
// ensures the returned data is as expected until the timeout occurs
func queryAndWait(t *testing.T, nodes Cluster, urlDb, q, expected, expectPattern string, timeout time.Duration) (string, bool, int) {
	v := url.Values{"q": []string{q}}
	if urlDb != "" {
		v.Set("db", urlDb)
	}

	sleep := 100 * time.Millisecond
	// Check to see if they set the env for duration sleep
	if sleepRaw := os.Getenv("TEST_SLEEP"); sleepRaw != "" {
		d, err := time.ParseDuration(sleepRaw)
		if err != nil {
			t.Fatal("Invalid TEST_SLEEP value:", err)
		}
		// this will limit the http log noise in the test output
		sleep = d
		timeout = d + 1
	}

	var done <-chan time.Time
	if timeout > 0 {
		timer := time.NewTimer(timeout)
		done = timer.C
		defer timer.Stop()
	}

	nQueriedOK := 0
	for {
		got, ok, n := query(t, nodes, urlDb, q, expected, expectPattern)
		if n > nQueriedOK {
			nQueriedOK = n
		}
		if ok {
			return got, ok, nQueriedOK
		}
		select {
		case <-done:
			return got, false, nQueriedOK
		case <-time.After(sleep):
		}
	}
}

// mergeMany ensures that when merging many series together and some of them have a different number
// of points than others in a group by interval the results are correct
var mergeMany = func(t *testing.T, node *TestNode, database, retention string) {
	for i := 1; i < 11; i++ {
		for j := 1; j < 5+i%3; j++ {
			data := fmt.Sprintf(`{"database": "%s", "retentionPolicy": "%s", "points": [{"name": "cpu", "timestamp": "%s", "tags": {"host": "server_%d"}, "fields": {"value": 22}}]}`,
				database, retention, time.Unix(int64(j), int64(0)).UTC().Format(time.RFC3339), i)
			write(t, node, data)
		}
	}
}

var limitAndOffset = func(t *testing.T, node *TestNode, database, retention string) {
	for i := 1; i < 10; i++ {
		data := fmt.Sprintf(`{"database": "%s", "retentionPolicy": "%s", "points": [{"name": "cpu", "timestamp": "%s", "tags": {"region": "us-east", "host": "server-%d"}, "fields": {"value": %d}}]}`,
			database, retention, time.Unix(int64(i), int64(0)).Format(time.RFC3339), i, i)
		write(t, node, data)
	}
}

func runTest_rawDataReturnsInOrder(t *testing.T, testName string, nodes Cluster, database, retention string, replicationFactor int) {
	// skip this test if they're just looking to run some of thd data tests
	if os.Getenv("TEST_PREFIX") != "" {
		return
	}
	t.Logf("Running %s:rawDataReturnsInOrder against %d-node cluster", testName, len(nodes))

	// Start by ensuring database and retention policy exist.
	createDatabase(t, testName, nodes, database)
	createRetentionPolicy(t, testName, nodes, database, retention, replicationFactor)
	numPoints := 500
	var expected string

	for i := 1; i < numPoints; i++ {
		data := fmt.Sprintf(`{"database": "%s", "retentionPolicy": "%s", "points": [{"name": "cpu", "timestamp": "%s", "tags": {"region": "us-east", "host": "server-%d"}, "fields": {"value": %d}}]}`,
			database, retention, time.Unix(int64(i), int64(0)).Format(time.RFC3339), i%10, i)
		write(t, nodes[0], data)
	}

	expected = fmt.Sprintf(`{"results":[{"series":[{"name":"cpu","columns":["time","count"],"values":[["1970-01-01T00:00:00Z",%d]]}]}]}`, numPoints-1)
	got, ok, nOK := queryAndWait(t, nodes, database, `SELECT count(value) FROM cpu`, expected, "", 120*time.Second)
	if !ok {
		t.Errorf("test %s:rawDataReturnsInOrder failed, SELECT count() query returned unexpected data\nexp: %s\n, got: %s\n%d nodes responded correctly", testName, expected, got, nOK)
		dumpClusterDiags(t, testName, nodes)
		dumpClusterStats(t, testName, nodes)
	}

	// Create expected JSON string dynamically.
	expectedValues := make([]string, 0)
	for i := 1; i < numPoints; i++ {
		expectedValues = append(expectedValues, fmt.Sprintf(`["%s",%d]`, time.Unix(int64(i), int64(0)).UTC().Format(time.RFC3339), i))
	}
	expected = fmt.Sprintf(`{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[%s]}]}]}`, strings.Join(expectedValues, ","))
	got, ok, nOK = query(t, nodes, database, `SELECT value FROM cpu`, expected, "")
	if !ok {
		t.Errorf("test %s failed, SELECT query returned unexpected data\nexp: %s\ngot: %s\n%d nodes responded correctly", testName, expected, got, nOK)
		dumpClusterDiags(t, testName, nodes)
		dumpClusterStats(t, testName, nodes)
	}
}

// runTests_Errors tests some basic error cases.
func runTests_Errors(t *testing.T, nodes Cluster) {
	t.Logf("Running tests against %d-node cluster", len(nodes))

	tests := []struct {
		name     string
		write    string // If equal to the empty string, no data is written.
		query    string // If equal to the blank string, no query is executed.
		expected string // If 'query' is equal to the blank string, this is ignored.
	}{
		{
			name:     "simple SELECT from non-existent database",
			write:    "",
			query:    `SELECT * FROM "qux"."bar".cpu`,
			expected: `{"results":[{"error":"database not found: qux"}]}`,
		},
	}

	for _, tt := range tests {
		if tt.write != "" {
			write(t, nodes[0], tt.write)
		}

		if tt.query != "" {
			got, ok, nOK := query(t, nodes, "", tt.query, tt.expected, "")
			if !ok {
				t.Errorf("Test '%s' failed, expected: %s, got: %s, %d nodes responded correctly", tt.name, tt.expected, got, nOK)
			}
		}
	}
}

// runTests tests write and query of data. Setting testNumbers allows only a subset of tests to be run.
func runTestsData(t *testing.T, testName string, nodes Cluster, database, retention string, replicationFactor int) {
	t.Logf("Running tests against %d-node cluster", len(nodes))

	yesterday := time.Now().Add(-1 * time.Hour * 24).UTC()
	now := time.Now().UTC()
	maxFloat64, _ := json.Marshal(math.MaxFloat64)

	// Start by ensuring database and retention policy exist.
	createDatabase(t, testName, nodes, database)
	createRetentionPolicy(t, testName, nodes, database, retention, replicationFactor)

	// The tests. Within these tests %DB% and %RP% will be replaced with the database and retention passed into
	// this function.
	tests := []struct {
		skip          bool    // Skip the test.
		reset         bool    // Delete and recreate the database.
		name          string  // Test name, for easy-to-read test log output.
		write         string  // If equal to the empty string, no data is written.
		writeFn       writeFn // If non-nil, called after 'write' data (if any) is written.
		query         string  // If equal to the blank string, no query is executed.
		queryDb       string  // If set, is used as the "db" query param.
		queryOne      bool    // If set, only 1 node is queried.
		expected      string  // If 'query' is equal to the blank string, this is ignored.
		expectPattern string  // Regexp alternative to expected field. (ignored if expected is set)
	}{
		// Data read and write tests
		{
			reset:    true,
			name:     "single point with timestamp",
			write:    `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "cpu", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": 100}}]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".cpu`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",100]]}]}]}`,
		},
		{
			name:     "single point count query with timestamp",
			query:    `SELECT count(value) FROM "%DB%"."%RP%".cpu`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]}]}]}`,
		},
		{
			name:     "single string point with timestamp",
			write:    `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "logs", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": "disk full"}}]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".logs`,
			expected: `{"results":[{"series":[{"name":"logs","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z","disk full"]]}]}]}`,
		},
		{
			name:     "single bool point with timestamp",
			write:    `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "status", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": "true"}}]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".status`,
			expected: `{"results":[{"series":[{"name":"status","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z","true"]]}]}]}`,
		},
		{
			name:     "single point, select with now()",
			query:    `SELECT * FROM "%DB%"."%RP%".cpu WHERE time < now()`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",100]]}]}]}`,
		},
		{
			name:     "single point, select with now(), two queries",
			query:    `SELECT * FROM "%DB%"."%RP%".cpu WHERE time < now();SELECT * FROM "%DB%"."%RP%".cpu WHERE time < now()`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",100]]}]},{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",100]]}]}]}`,
		},

		{
			name:          "measurement not found",
			query:         `SELECT value FROM "%DB%"."%RP%".foobarbaz`,
			expectPattern: `.*measurement not found.*`,
		},
		{
			name:     "field not found",
			query:    `SELECT abc FROM "%DB%"."%RP%".cpu WHERE time < now()`,
			expected: `{"results":[{"error":"unknown field or tag name in select clause: abc"}]}`,
		},
		{
			name:     "empty result",
			query:    `SELECT value FROM cpu WHERE time >= '3000-01-01 00:00:05'`,
			queryDb:  "%DB%",
			expected: `{"results":[{}]}`,
		},

		{
			reset: true,
			name:  "two points with timestamp",
			write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "cpu", "timestamp": "2015-02-28T01:04:36.703820946Z", "fields": {"value": 100}},
		                                                                         {"name": "cpu", "timestamp": "2015-02-28T01:03:36.703820946Z", "fields": {"value": 100}}
		                                                                        ]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".cpu`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",100],["2015-02-28T01:04:36.703820946Z",100]]}]}]}`,
		},

		{
			reset: true,
			name:  "two points with negative values",
			write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "cpu", "timestamp": "2015-02-28T01:04:36.703820946Z", "fields": {"value": -200}},
                                                                                 {"name": "cpu", "timestamp": "2015-02-28T01:03:36.703820946Z", "fields": {"value": -100}}
                                                                                ]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".cpu`,
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",-100],["2015-02-28T01:04:36.703820946Z",-200]]}]}]}`,
		},

		// Data read and write tests using relative time
		{
			reset:    true,
			name:     "single point with timestamp pre-calculated for past time queries yesterday",
			write:    `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "cpu", "timestamp": "` + yesterday.Format(time.RFC3339Nano) + `", "tags": {"host": "server01"}, "fields": {"value": 100}}]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".cpu where time >= '` + yesterday.Add(-1*time.Minute).Format(time.RFC3339Nano) + `'`,
			expected: fmt.Sprintf(`{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["%s",100]]}]}]}`, yesterday.Format(time.RFC3339Nano)),
		},
		{
			reset:    true,
			name:     "single point with timestamp pre-calculated for relative time queries now",
			write:    `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "cpu", "timestamp": "` + now.Format(time.RFC3339Nano) + `", "tags": {"host": "server01"}, "fields": {"value": 100}}]}`,
			query:    `SELECT * FROM "%DB%"."%RP%".cpu where time >= now() - 1m`,
			expected: fmt.Sprintf(`{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[["%s",100]]}]}]}`, now.Format(time.RFC3339Nano)),
		},

		// Merge tests.
		{
			reset:    true,
			name:     "merge many",
			writeFn:  mergeMany,
			query:    `SELECT count(value) FROM cpu WHERE time >= '1970-01-01T00:00:01Z' AND time <= '1970-01-01T00:00:06Z' GROUP BY time(1s)`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","count"],"values":[["1970-01-01T00:00:01Z",10],["1970-01-01T00:00:02Z",10],["1970-01-01T00:00:03Z",10],["1970-01-01T00:00:04Z",10],["1970-01-01T00:00:05Z",7],["1970-01-01T00:00:06Z",3]]}]}]}`,
		},

		// Limit and offset
		{
			reset:    true,
			name:     "limit and offset",
			writeFn:  limitAndOffset,
			query:    `SELECT count(value) FROM cpu GROUP BY * SLIMIT 2 SOFFSET 1`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu","tags":{"host":"server-2","region":"us-east"},"columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]},{"name":"cpu","tags":{"host":"server-3","region":"us-east"},"columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]}]}]}`,
		},
		{
			query:    `SELECT count(value) FROM cpu GROUP BY * SLIMIT 2 SOFFSET 3`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu","tags":{"host":"server-4","region":"us-east"},"columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]},{"name":"cpu","tags":{"host":"server-5","region":"us-east"},"columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]}]}]}`,
		},
		{
			query:    `SELECT count(value) FROM cpu GROUP BY * SLIMIT 3 SOFFSET 8`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu","tags":{"host":"server-9","region":"us-east"},"columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]}]}]}`,
		},

		// FROM regex tests
		{
			reset: true,
			name:  "FROM regex using default db and rp",
			write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [
				{"name": "cpu1", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": 10}},
				{"name": "cpu2", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": 20}},
				{"name": "cpu3", "timestamp": "2015-02-28T01:03:36.703820946Z", "tags": {"host": "server01"}, "fields": {"value": 30}}
			]}`,
			query:    `SELECT * FROM /cpu[13]/`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu1","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",10]]},{"name":"cpu3","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",30]]}]}]}`,
		},
		{
			name:     `FROM regex specifying db and rp`,
			query:    `SELECT * FROM "%DB%"."%RP%"./cpu[13]/`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu1","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",10]]},{"name":"cpu3","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",30]]}]}]}`,
		},
		{
			name:     `FROM regex using specified db and default rp`,
			query:    `SELECT * FROM "%DB%"../cpu[13]/`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu1","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",10]]},{"name":"cpu3","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",30]]}]}]}`,
		},
		{
			name:     `FROM regex using default db and specified rp`,
			query:    `SELECT * FROM "%RP%"./cpu[13]/`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu1","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",10]]},{"name":"cpu3","columns":["time","value"],"values":[["2015-02-28T01:03:36.703820946Z",30]]}]}]}`,
		},

		// Aggregations
		{
			reset: true,
			name:  "large mean",
			write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [
				{"name": "cpu", "timestamp": "2015-04-20T14:27:40Z", "fields": {"value": ` + string(maxFloat64) + `}},
				{"name": "cpu", "timestamp": "2015-04-20T14:27:41Z", "fields": {"value": ` + string(maxFloat64) + `}}
			]}`,
			query:    `SELECT mean(value) FROM cpu`,
			queryDb:  "%DB%",
			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","mean"],"values":[["1970-01-01T00:00:00Z",` + string(maxFloat64) + `]]}]}]}`,
		},
		{
			reset: true,
			name:  "aggregations",
			write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [
				{"name": "cpu", "timestamp": "2000-01-01T00:00:00Z", "tags": {"region": "us-east"}, "fields": {"value": 20}},
{'fraction_non_alphanumeric': 0.2054290646062798, 'fraction_numerical': 0.047714943284563535, 'mean_word_length': 4.07085547566948, 'pattern_counts': {'":': 524, '<': 33, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4876907', 'n_tokens_mistral': 19371, 'n_tokens_neox': 16752, 'n_words': 4370}