//go:build !kafka_sasl
// +build !kafka_sasl

package kafka

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"math/rand"
	"net/http"
	"os"
	"reflect"
	"sort"
	"strconv"
	"strings"
	"testing"
	"time"

	confluent "github.com/confluentinc/confluent-kafka-go/kafka"
	pilosaclient "github.com/featurebasedb/featurebase/v3/client"
	"github.com/featurebasedb/featurebase/v3/idk"
	"github.com/featurebasedb/featurebase/v3/idk/kafka/csrc"
	"github.com/featurebasedb/featurebase/v3/logger"
	liavro "github.com/linkedin/goavro/v2"
)

var (
	pilosaHost     string
	pilosaTLSHost  string
	pilosaGrpcHost string
	kafkaHost      string
	registryHost   string
	certPath       string
)

func init() {
	local := false
	var ok bool
	if pilosaHost, ok = os.LookupEnv("IDK_TEST_PILOSA_HOST"); !ok {
		if local {
			pilosaHost = "localhost:10101"
		} else {
			pilosaHost = "pilosa:10101"
		}
	}
	if pilosaTLSHost, ok = os.LookupEnv("IDK_TEST_PILOSA_TLS_HOST"); !ok {
		pilosaTLSHost = "https://pilosa-tls:10111"
	}
	if pilosaGrpcHost, ok = os.LookupEnv("IDK_TEST_PILOSA_GRPC_HOST"); !ok {
		if local {
			pilosaGrpcHost = "localhost:20101"
		} else {
			pilosaGrpcHost = "pilosa:20101"
		}
	}
	if kafkaHost, ok = os.LookupEnv("IDK_TEST_KAFKA_HOST"); !ok {
		if local {
			kafkaHost = "localhost:9092"
		} else {
			kafkaHost = "kafka:9092"
		}
	}
	if registryHost, ok = os.LookupEnv("IDK_TEST_REGISTRY_HOST"); !ok {
		if local {
			registryHost = "localhost:8081"
		} else {
			registryHost = "schema-registry:8081"
		}
	}
	if certPath, ok = os.LookupEnv("IDK_TEST_CERT_PATH"); !ok {
		certPath = "/certs"
	}
}

func configureTestFlags(main *Main) {
	main.PilosaHosts = []string{pilosaHost}
	main.PilosaGRPCHosts = []string{pilosaGrpcHost}
	main.KafkaBootstrapServers = []string{kafkaHost}
	main.SchemaRegistryURL = registryHost
	main.Verbose = true
	// intentionally low timeout — if this gets triggered it shouldn't
	// have any negative effects
	main.Timeout = time.Millisecond * 20
	main.Stats = ""
	_, main.Verbose = os.LookupEnv("IDK_TEST_VERBOSE")
}

func TestCmdAutoID(t *testing.T) {
	t.Parallel()

	if testing.Short() {
		t.Skip()
	}

	rand.Seed(time.Now().UnixNano())
	topic := strconv.Itoa(rand.Int())

	fields := []string{"val"}
	records := [][]interface{}{
		{2},
		{3},
		{4},
		{6},
	}

	m, err := NewMain()
	if err != nil {
		t.Fatalf("creating main %v", err)
	}
	configureTestFlags(m)
	m.Index = fmt.Sprintf("cmd_test_autoid239ij%s", topic)
	m.BatchSize = 2 // need to test at a batch size less than the # of records, greater than, and equal to
	m.Topics = []string{topic}
	m.MaxMsgs = uint64(len(records))
	m.AutoGenerate = true
	m.ExternalGenerate = true

	// load schema
	licodec := liDecodeTestSchema(t, "ids.json")
	schemaID := postSchema(t, "ids.json", "ids", m.SchemaRegistryURL, nil)
	p, err := confluent.NewProducer(&confluent.ConfigMap{
		"bootstrap.servers": kafkaHost,
	})
	if err != nil {
		t.Fatalf("Failed to create producer: %s", err)
	}
	defer p.Close()
	// put records in kafka
	tCreateTopic(t, topic, p)

	msgs := make([]map[string]interface{}, len(records))
	for i, vals := range records {
		msgs[i] = makeRecord(t, fields, vals)
	}
	tPutRecordsKafka(t, p, topic, schemaID, licodec, "a", msgs...)

	err = m.Run()
	if err != nil {
		t.Fatalf("running main: %v", err)
	}

	client := m.PilosaClient()
	schema, err := client.Schema()
	if err != nil {
		t.Fatalf("getting client: %v", err)
	}
	index := schema.Index(m.Index)
	defer func() {
		err := client.DeleteIndex(index)
		if err != nil {
			t.Logf("deleting index: %v", err)
		}
	}()

	qr, err := client.Query(index.Count(index.All()))
	if err != nil {
		t.Errorf("querying: %v", err)
	}
	if qr.Result().Count() != 4 {
		t.Errorf("wrong count for columns, %d is not 4", qr.Result().Count())
	}

	qr, err = client.Query(pilosaclient.NewPQLBaseQuery(`Count(Distinct(All(), field="val"))`, index, nil))
	if err != nil {
		t.Errorf("querying: %v", err)
	}
	if qr.Result().Count() != 4 {
		t.Errorf("wrong count for val, %d is not 4", qr.Result().Count())
	}
}

func TestConfigOptions(t *testing.T) {
	m, err := NewMain()
	if err != nil {
		t.Fatalf("getting main: %v", err)
	}

	debug := "consumer"
	clientID := "blah"
	autoOffsetReset := true
	groupInstanceID := "instId"
	maxPollIntervalMs := 30000
	sessionTimeoutMs := 20000
	socketKeepaliveEnable := "true"

	m.KafkaDebug = debug
	m.KafkaClientId = clientID
	m.SkipOld = autoOffsetReset
	m.KafkaGroupInstanceId = groupInstanceID
	m.KafkaMaxPollInterval = maxPollIntervalMs
	m.KafkaSessionTimeout = sessionTimeoutMs
	m.KafkaSocketKeepaliveEnable = socketKeepaliveEnable

	//kafka config name, expected value
	tests := [][2]interface{}{
		{"debug", debug},
		{"client.id", clientID},
		{"auto.offset.reset", "latest"},
		{"group.instance.id", groupInstanceID},
		{"max.poll.interval.ms", maxPollIntervalMs},
		{"session.timeout.ms", sessionTimeoutMs},
		{"socket.keepalive.enable", socketKeepaliveEnable},
	}

	// call NewSource to get and open source which calls some setup
	// stuff that we want. We don't expect it to open successfully
	// though.
	src, err := m.NewSource()
	if err != nil {
		t.Fatalf("error calling new source: %v", err)
	}
	defer src.Close()

	cfg := src.(*Source).ConfigMap

	for _, pair := range tests {
		if val, err := cfg.Get(pair[0].(string), nil); err != nil || fmt.Sprintf("%v", val) != fmt.Sprintf("%v", pair[1]) {
			t.Fatalf("unexpected val for %s val: %v, err: %v", pair[0].(string), val, err)
		}
	}
}

func TestCmdMainOne(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.Skip()
	}

	tests := []struct {
		name             string
		PrimaryKeyFields []string
		IDField          string
		PilosaHosts      []string
		RegistryURL      string
		TLS              *idk.TLSConfig
		expRhinoKeys     []string
		expRhinoCols     []uint64
	}{
		{
			name:             "3 primary keys str/str/int",
			PrimaryKeyFields: []string{"abc", "db", "user_id"},
			expRhinoKeys:     []string{"2|1|159", "4|3|44", "123456789|q2db_1234|432"}, // "2" + "1" + uint32(159)

		},
		{
			name:             "3 primary keys str/str/int TLS",
			PrimaryKeyFields: []string{"abc", "db", "user_id"},
			PilosaHosts:      []string{pilosaTLSHost},
			TLS: &idk.TLSConfig{
				CertificatePath:          certPath + "/theclient.crt",
				CertificateKeyPath:       certPath + "/theclient.key",
				CACertPath:               certPath + "/ca.crt",
				EnableClientVerification: true,
			},
			expRhinoKeys: []string{"2|1|159", "4|3|44", "123456789|q2db_1234|432"}, // "2" + "1" + uint32(159)

		},
		{
			name:         "IDField int",
			IDField:      "user_id",
			expRhinoCols: []uint64{44, 159, 432},
		},
	}

	rand := rand.New(rand.NewSource(time.Now().UnixNano()))

	for _, test := range tests {
		test := test
		a := rand.Int()
		t.Run(test.name, func(t *testing.T) {
			t.Parallel()

			fields := []string{"abc", "db", "user_id", "all_users", "has_deleted_date", "central_group", "custom_audiences", "desktop_boolean", "desktop_frequency", "desktop_recency", "product_boolean_historical_forestry_cravings_or_bugles", "ddd_category_total_current_rhinocerous_checking", "ddd_category_total_current_rhinocerous_thedog_cheetah", "survey1234", "days_since_last_logon", "elephant_added_for_account"}

			records := [][]interface{}{
				{"2", "1", 159, map[string]interface{}{"boolean": true}, map[string]interface{}{"boolean": false}, map[string]interface{}{"string": "cgr"}, map[string]interface{}{"array": []string{"a", "b"}}, nil, map[string]interface{}{"int": 7}, nil, nil, map[string]interface{}{"float": 5.4}, nil, map[string]interface{}{"org.test.survey1234": "yes"}, map[string]interface{}{"float": 8.0}, nil},
				{"4", "3", 44, map[string]interface{}{"boolean": true}, map[string]interface{}{"boolean": false}, map[string]interface{}{"string": "cgr"}, map[string]interface{}{"array": []string{"a", "b"}}, nil, map[string]interface{}{"int": 7}, nil, nil, map[string]interface{}{"float": 5.4}, nil, map[string]interface{}{"org.test.survey1234": "yes"}, map[string]interface{}{"float": 8.0}, nil},
				{"123456789", "q2db_1234", 432, map[string]interface{}{"boolean": false}, map[string]interface{}{"boolean": false}, map[string]interface{}{"string": "cgr"}, map[string]interface{}{"array": []string{"a", "b"}}, nil, map[string]interface{}{"int": 7}, nil, nil, map[string]interface{}{"float": 5.9}, nil, map[string]interface{}{"org.test.survey1234": "yes"}, map[string]interface{}{"float": 8.0}, nil},
				{"123456789", "q2db_1234", 432, map[string]interface{}{"boolean": false}, map[string]interface{}{"boolean": false}, map[string]interface{}{"string": "cgr"}, map[string]interface{}{"array": []string{"a", "b"}}, nil, map[string]interface{}{"int": 7}, nil, nil, map[string]interface{}{"float": 5.4}, nil, map[string]interface{}{"org.test.survey1234": "yes"}, map[string]interface{}{"float": 8.0}, nil},
				{"2", "1", 159, map[string]interface{}{"boolean": false}, map[string]interface{}{"boolean": false}, map[string]interface{}{"string": "cgr"}, map[string]interface{}{"array": []string{"a", "b"}}, nil, map[string]interface{}{"int": 7}, nil, nil, map[string]interface{}{"float": 5.4}, nil, map[string]interface{}{"org.test.survey1234": "yes"}, map[string]interface{}{"float": 8.0}, nil},
				{"4", "3", 44, map[string]interface{}{"boolean": false}, map[string]interface{}{"boolean": false}, map[string]interface{}{"string": "cgr"}, map[string]interface{}{"array": []string{"a", "b"}}, nil, map[string]interface{}{"int": 7}, nil, nil, map[string]interface{}{"float": 5.4}, nil, map[string]interface{}{"org.test.survey1234": "yes"}, map[string]interface{}{"float": 8.0}, nil},
				{"4", "3", 44, map[string]interface{}{"boolean": true}, map[string]interface{}{"boolean": false}, map[string]interface{}{"string": "cgr"}, map[string]interface{}{"array": []string{"a", "b"}}, nil, map[string]interface{}{"int": 7}, nil, nil, map[string]interface{}{"float": 5.4}, nil, map[string]interface{}{"org.test.survey1234": "yes"}, map[string]interface{}{"float": 8.0}, nil},
				{"4", "3", 44, nil, map[string]interface{}{"boolean": false}, map[string]interface{}{"string": "cgr"}, map[string]interface{}{"array": []string{"a", "b"}}, nil, map[string]interface{}{"int": 7}, nil, nil, map[string]interface{}{"float": 5.4}, nil, map[string]interface{}{"org.test.survey1234": "yes"}, map[string]interface{}{"float": 8.0}, nil}, // send all_users nil - regression test to make sure it isn't cleared by the nil
			}

			topic := strconv.Itoa(a)

			// create Main and run with MaxMsgs
			m, err := NewMain()
			if err != nil {
				t.Fatalf("creating main %v", err)
			}
			configureTestFlags(m)
			m.Index = fmt.Sprintf("cmd_test_index239ij%s", topic)
			m.PrimaryKeyFields = test.PrimaryKeyFields
			m.IDField = test.IDField
			m.PackBools = "bools"
			m.BatchSize = 3 // need to test at a batch size less than the # of records, greater than, and equal to
			m.Topics = []string{topic}
			m.MaxMsgs = uint64(len(records))
			if test.PilosaHosts != nil {
				m.PilosaHosts = test.PilosaHosts
			}
			if test.TLS != nil {
				m.TLS = *test.TLS
			}
			if test.RegistryURL != "" {
				m.SchemaRegistryURL = test.RegistryURL
			}

			// load big schema
			licodec := liDecodeTestSchema(t, "bigschema.json")
			tlsConf, err := idk.GetTLSConfig(test.TLS, logger.NopLogger)
			if err != nil {
				t.Fatalf("getting tls config: %v", err)
			}
			schemaID := postSchema(t, "bigschema.json", "bigschema2", m.SchemaRegistryURL, tlsConf)

			p, err := confluent.NewProducer(&confluent.ConfigMap{
				"bootstrap.servers": kafkaHost,
			})
			if err != nil {
				t.Fatalf("Failed to create producer: %s", err)
			}
			defer p.Close()
			// put records in kafka
			tCreateTopic(t, topic, p)

			msgs := make([]map[string]interface{}, len(records))
			for i, vals := range records {
				msgs[i] = makeRecord(t, fields, vals)
			}
			tPutRecordsKafka(t, p, topic, schemaID, licodec, "akey", msgs...)
			err = m.Run()
			if err != nil {
				t.Fatalf("running main: %v", err)
			}

			client := m.PilosaClient()
			schema, err := client.Schema()
			if err != nil {
				t.Fatalf("getting client: %v", err)
			}
			index := schema.Index(m.Index)
			defer func() {
				err := client.DeleteIndex(index)
				if err != nil {
					t.Logf("deleting index: %v", err)
				}
			}()

			if status, body, err := client.HTTPRequest("POST", "/recalculate-caches", nil, nil); err != nil {
				t.Fatalf("recalculating cache: status: %d, response: %s, error: %s", status, body, err)
			}

			// check data in Pilosa
			if !index.HasField("abc") {
				t.Fatal("don't have abc")
			}
			abc := index.Field("abc")
			qr, err := client.Query(index.Count(abc.Row("2")))
			if err != nil {
				t.Errorf("querying: %v", err)
			}
			if qr.Result().Count() != 1 {
				t.Errorf("wrong count for abc, %d is not 1", qr.Result().Count())
			}

			bools := index.Field("bools")
			qr, err = client.Query(bools.TopN(10))
			if err != nil {
				t.Fatalf("querying: %v", err)
			}
			ci := sortableCRI(qr.Result().CountItems())
			exp := sortableCRI{{Count: 1, Key: "all_users"}}
			sort.Sort(ci)
			sort.Sort(exp)
			if !reflect.DeepEqual(ci, exp) {
				qr, err = client.Query(bools.Row("all_users"))
				if err != nil {
					t.Fatalf("querying: %v", err)
				}
				t.Errorf("unexpected result exp/got\n%+v\n%+v\nall_users:%+v", exp, ci, qr.Result().Row().Keys)
			}

			qr, err = client.Query(bools.Row("all_users"))
			if err != nil {
				t.Fatalf("querying: %v", err)
			}
			if test.IDField == "" {
				if keys := qr.Result().Row().Keys; !reflect.DeepEqual(keys, []string{"4|3|44"}) {
					t.Errorf("unexpected keys %v", keys)
				}
			} else if cols := qr.Result().Row().Columns; !reflect.DeepEqual(cols, []uint64{44}) {
				t.Errorf("unexpected cols %v", cols)
			}

			bools = index.Field("bools-exists")
			qr, err = client.Query(bools.TopN(10))
			if err != nil {
				t.Fatalf("querying: %v", err)
			}
			ci = sortableCRI(qr.Result().CountItems())
			exp = sortableCRI{{Count: 3, Key: "all_users"}, {Count: 3, Key: "has_deleted_date"}}
			sort.Sort(ci)
			sort.Sort(exp)
			if !reflect.DeepEqual(ci, exp) {
				t.Errorf("unexpected result exp/got\n%+v\n%+v", exp, ci)
			}

			rhino := index.Field("ddd_category_total_current_rhinocerous_checking")
			qr, err = client.Query(rhino.Between(5, 6))
			if err != nil {
				t.Fatalf("querying: %v", err)
			}
			keys := qr.Result().Row().Keys
			sort.Strings(keys)
			sort.Strings(test.expRhinoKeys)
			if test.expRhinoKeys != nil {
				if !reflect.DeepEqual(keys, test.expRhinoKeys) {
					t.Errorf("wrong keys: %v, exp: %v", keys, test.expRhinoKeys)
				}
			}
			if test.expRhinoCols != nil {
				bod := tDoHTTPPost(t, fmt.Sprintf("http://"+pilosaHost+"/index/%s/query", m.Index), "application/pql", "Row(ddd_category_total_current_rhinocerous_checking==5.4)")
				if !strings.Contains(bod, "44,159,432") {
					t.Errorf("unexpected result with float query: '%s'", bod)
				}
				if cols := qr.Result().Row().Columns; !reflect.DeepEqual(cols, test.expRhinoCols) {
					t.Errorf("wrong cols: %v, exp: %v", cols, test.expRhinoCols)
				}
			}

			if !index.HasField("survey1234") {
				t.Fatalf("don't have survey1234")
			}
			survey1234 := index.Field("survey1234")
			if typ := survey1234.Opts().Type(); typ != pilosaclient.FieldTypeMutex {
				t.Errorf("survey1234 of unexpected type: %v", typ)
			}
		})
	}
}

func TestCmdSourceTimeout(t *testing.T) {
	//	t.Parallel()

	rand.Seed(time.Now().UnixNano())
	topic := strconv.Itoa(rand.Int())

	// create Main and run with MaxMsgs
	m, err := NewMain()
	if err != nil {
		t.Fatalf("creating main %v", err)
	}
	configureTestFlags(m)
	m.Index = fmt.Sprintf("cmd_test_schemachange239ij%s", topic)
	m.IDField = "id"
	m.PackBools = "bools"
	m.BatchSize = 5
	m.Topics = []string{topic}
	m.Timeout = time.Millisecond * 20
	m.MaxMsgs = uint64(3)

	schemaStr1 := `{"type": "record","namespace": "cmdSrcTimeout","name": "basic","fields": [{"name":"id","type":"long"},{"name":"a","type": "string"}]}`
	licodec1, err := liavro.NewCodec(schemaStr1)
	if err != nil {
		t.Fatalf("li parsing schema: %v", err)
	}

	schemaClient := csrc.NewClient(m.SchemaRegistryURL, nil, nil)
	resp, err := schemaClient.PostSubjects("cmdSourceTimeout", schemaStr1)
	if err != nil {
		t.Fatalf("posting schema: %v", err)
	}
	schemaID1 := resp.ID

	p, err := confluent.NewProducer(&confluent.ConfigMap{
		"bootstrap.servers": kafkaHost,
	})
	if err != nil {
		t.Fatalf("Failed to create producer: %s", err)
	}
	defer p.Close()

	rec := makeRecord(t, []string{"id", "a"}, []interface{}{1, "v"})
	tPutRecordsKafka(t, p, topic, schemaID1, licodec1, "akey", rec)
	rec = makeRecord(t, []string{"id", "a"}, []interface{}{2, "v"})
	tPutRecordsKafka(t, p, topic, schemaID1, licodec1, "akey", rec)

	// create client manually and then create schema. We have to do
	// this ahead of time because we're calling m.Run concurrently.
	client, err := pilosaclient.NewClient(m.PilosaHosts)
	if err != nil {
		t.Fatalf("creating new client: %v", err)
	}
	schema, err := client.Schema()
	if err != nil {
		t.Fatalf("getting client: %v", err)
	}
	index := schema.Index(m.Index)
	af := index.Field("a", pilosaclient.OptFieldKeys(true))
	if err := client.SyncIndex(index); err != nil {
		t.Fatalf("syncing index: %v", err)
	}
	defer func() {
		if err := client.DeleteIndexByName(m.Index); err != nil {
			t.Fatalf("deleting index: %s", m.Index)
		}
	}()

	// Run ingest in a goroutine so we can see the effect of the
	// timeout flush.
	signal := make(chan struct{})
	go func() {
		err = m.Run()
		if err != nil {
			t.Logf("running main: %v", err)
		}
		close(signal)
	}()

	// query Pilosa repeatedly until we see the two records reflected there.
	for {
		qr, err := client.Query(index.Count(af.Row("v")))
		if err != nil {
			t.Fatalf("querying: %v", err)
		}
		if cnt := qr.Result().Count(); cnt == 2 {
			break
		} else if cnt == 0 {
			time.Sleep(time.Millisecond * 50)
			continue
		} else {
			t.Fatalf("unexpected count: %v", cnt)
		}
	}

	// put another message to make m.Run finish (MaxMsgs = 3)
	tPutRecordsKafka(t, p, topic, schemaID1, licodec1, "akey", rec)

	// make sure the Run goroutine actually finished
	<-signal
}

func TestCmdSchemaChange(t *testing.T) {
	t.Parallel()

	rand.Seed(time.Now().UnixNano())
	topic := strconv.Itoa(rand.Int())

	// create Main and run with MaxMsgs
	m, err := NewMain()
	if err != nil {
		t.Fatalf("creating main %v", err)
	}
	configureTestFlags(m)
	m.Index = fmt.Sprintf("cmd_test_schemachange239ij%s", topic)
	m.IDField = "id"
	m.PackBools = "bools"
	m.BatchSize = 1
	m.Topics = []string{topic}
	m.MaxMsgs = uint64(3)

	// load two schemas
	schemaStr1 := `{"type": "record","namespace": "c.e","name": "F","fields": [{"name":"id","type":"long"},{"name":"a","type": "boolean"},{"name": "c", "type": "int"}]}`
	licodec1, err := liavro.NewCodec(schemaStr1)
	if err != nil {
		t.Fatalf("li parsing schema: %v", err)
	}
	schemaStr2 := `{"type": "record","namespace": "c.e","name": "F","fields": [{"name":"id","type":"long"},{"name":"a","type": "boolean"},{"name":"b","type":"long"},{"name": "c", "type": "int"}]}`
	licodec2, err := liavro.NewCodec(schemaStr2)
	if err != nil {
		t.Fatalf("li parsing schema: %v", err)
	}

	// have to change schema name to change field from int to long
	schemaStr3 := `{"type": "record","namespace": "c.e","name": "F2","fields": [{"name":"id","type":"long"},{"name":"a","type": "boolean"},{"name":"b","type":"long"},{"name": "c", "type": "long"}]}`
	licodec3, err := liavro.NewCodec(schemaStr3)
	if err != nil {
		t.Fatalf("li parsing schema: %v", err)
	}

	schemaClient := csrc.NewClient(m.SchemaRegistryURL, nil, nil)
	resp, err := schemaClient.PostSubjects("subj", schemaStr1)
	if err != nil {
		t.Fatalf("posting schema: %v", err)
	}
	schemaID1 := resp.ID
	resp, err = schemaClient.PostSubjects("subj2", schemaStr2)
	if err != nil {
		t.Fatalf("posting schema: %v", err)
	}
	schemaID2 := resp.ID
	resp, err = schemaClient.PostSubjects("subj3", schemaStr3)
	if err != nil {
		t.Fatalf("posting schema: %v", err)
	}
	schemaID3 := resp.ID

	p, err := confluent.NewProducer(&confluent.ConfigMap{
		"bootstrap.servers": kafkaHost,
	})
	if err != nil {
		t.Fatalf("Failed to create producer: %s", err)
	}
	defer p.Close()

	tCreateTopic(t, topic, p)

	rec := makeRecord(t, []string{"id", "a", "c"}, []interface{}{1, true, 7})
	tPutRecordsKafka(t, p, topic, schemaID1, licodec1, "akey", rec)
	rec = makeRecord(t, []string{"id", "a", "b", "c"}, []interface{}{2, true, 22, 8})
	tPutRecordsKafka(t, p, topic, schemaID2, licodec2, "akey", rec)
	rec = makeRecord(t, []string{"id", "a", "b", "c"}, []interface{}{2, true, 22, 9_876_543_210})
	tPutRecordsKafka(t, p, topic, schemaID3, licodec3, "akey", rec)

	// run ingest
	err = m.Run()
	if err != nil {
		t.Logf("running main: %v", err)
	}

	client := m.PilosaClient()
	schema, err := client.Schema()
	if err != nil {
		t.Fatalf("getting client: %v", err)
	}
	index := schema.Index(m.Index)
	defer func() {
		err := client.DeleteIndex(index)
		if err != nil {
			t.Logf("deleting index: %v", err)
		}
	}()

	if status, body, err := client.HTTPRequest("POST", "/recalculate-caches", nil, nil); err != nil {
		t.Fatalf("recalculating cache: status: %d, response: %s, error: %s", status, body, err)
	}

	fieldA := index.Field("bools")
	fieldB := index.Field("b")

	qresp, err := client.Query(index.BatchQuery(fieldA.Row("a"), fieldB.GTE(0)))
	if err != nil {
		t.Fatalf("querying: %v", err)
	}
	aRow := qresp.Results()[0].Row().Columns
	if !reflect.DeepEqual(aRow, []uint64{1, 2}) {
		t.Errorf("unexpected columns for a: %v", aRow)
	}
	bRow := qresp.Results()[1].Row().Columns
	if !reflect.DeepEqual(bRow, []uint64{2}) {
		t.Errorf("unexpected columns for a: %v", bRow)
	}

	qresp, err = client.Query(index.RawQuery("Row(c==7)"))
	if err != nil {
		t.Fatalf("querying: %v", err)
	}
	sevenRow := qresp.Results()[0].Row().Columns
	if !reflect.DeepEqual(sevenRow, []uint64{1}) {
		t.Errorf("unexpected columns for 8: %v", sevenRow)
	}

	qresp, err = client.Query(index.RawQuery("Row(c==9876543210)"))
	if err != nil {
		t.Fatalf("querying: %v", err)
	}
	nineBRow := qresp.Results()[0].Row().Columns
	if !reflect.DeepEqual(nineBRow, []uint64{2}) {
		t.Errorf("unexpected columns for 9876543210: %v", nineBRow)
	}
}

type ConsumerTestConfig struct {
	idType    string   // must be "generated", "id", or "string"
	keyFields []string // field used for "id" or "string" record keys
	topic     string
	delete    bool
}

/*
Struct that captures data required to run a test that meets the following conditions:
  - There is a JSON file will plain text records to write to kafka
  - There is an Avro Schema that will be used to record records
    above before writing them to kafka
  - The data written above will be consumed and set to FeatureBase
  - There is a list of PQL queries that can be used to determine if
    above went correctly or incorrectly

For test that require multiple iterations of " consumer data from kafka, write it to
FeatureBase, and run queries against it, "
*/
type ConsumerTest struct {
	name              string
	pathsToAvroSchema []string // code currently prepends values with ./testdata/schema/
	pathsToRecords    []string
	consumerConfigs   []ConsumerTestConfig
	index             string
	queries           [][]string
	expectedResults   [][]string
	pilosaHosts       string
	kafkaHost         string
	registryURL       string
}

func TestAddingRemovingData(t *testing.T) {
	//t.Parallel()
	/*
		at a high level, a test here represents
		  - an avro schema
		  - a set of records to ingest to kafka
		  - an ingest configuration
		  - query to run to confirm the data was ingest properly

		see test
	*/
	tests := []ConsumerTest{

		{ // confirm time quantums are being ingested
			name:              "time quantums exist",
			pathsToAvroSchema: []string{"timeQuantum.json"},
			pathsToRecords:    []string{"./testdata/records/timeQuantum.json"},
			pilosaHosts:       pilosaHost,
			registryURL:       registryHost,
			kafkaHost:         kafkaHost,
			consumerConfigs: []ConsumerTestConfig{
				{
					idType:    "string",
					keyFields: []string{"device"}, //can be multiple for idType: string but a single value otherwise
					topic:     "timequantums",
					delete:    false,
				},
			},
			index: "timequantums",
			queries: [][]string{
				{
					"Row(segment_ts='7R83')",
					"Row(segment_ts='7R83', from=\"2023-02-17T00:00\", to=\"2023-02-18T00:00\")",
					"Row(segment_ts='7R83', from=\"2023-02-16T00:00\", to=\"2023-02-17T00:00\")",
				},
			},
			expectedResults: [][]string{
				{
					"{\"results\":[{\"columns\":[],\"keys\":[\"0QKtSTqJYXMZWvVe\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"0QKtSTqJYXMZWvVe\"]}]}\n",
					"{\"results\":[{\"columns\":[]}]}\n",
				},
			},
		},
		{ // confirm all types are being ingest as they should
			name:              "all values can be inserted",
			pathsToAvroSchema: []string{"alltypes.json"},
			pathsToRecords:    []string{"./testdata/records/alltypes.json"},
			pilosaHosts:       pilosaHost,
			registryURL:       registryHost,
			kafkaHost:         kafkaHost,
			consumerConfigs: []ConsumerTestConfig{
				{
					idType:    "string",
					keyFields: []string{"pk0", "pk1", "pk2"},
					topic:     "alltypes",
					delete:    false,
				},
			},
			index: "alltypes",
			queries: [][]string{
				{
					"Count(All())",
					"Count(ConstRow(columns=['u2Yr4|sHaUv|x5z8P', 'DY2Ui|kUbdU|pjxqm']))",
					"Count(Row(stringset_string='58KIR'))",
					"Count(Row(string_string='8MGwy'))",
					"Count(Row(stringtq_string='ivWWb'))",
					"Count(Row(stringtq_string='ivWWb', to=\"2023-02-03\"))",
					"Count(Row(stringtq_string='ivWWb', from=\"2023-02-03\", to=\"2023-02-04\"))",
					"Count(Union(Row(stringset_bytes='eNKWF'),Row(stringset_bytes='5ptDx')))",
					"Row(string_bytes='vTwn4')",
					"Row(stringsettq_bytes='798ka')",
					"Row(stringsettq_bytes='798ka', from=\"2023-02-18\")",
					"Row(stringsettq_bytes='798ka', from=\"2023-02-16\", to=\"2023-02-18\")",
					"Intersect(Row(stringset_stringarray='u2Yr4'), Row(stringset_stringarray='PYE8V'), Row(stringset_stringarray='VBcyJ'), Row(stringset_stringarray='Chgzr'), Row(stringset_stringarray='DY2Ui'))",
					"Row(stringtq_stringarray='oxjI0', from=\"2023-01-29\", to=\"2023-01-31\")",
					"Intersect(Row(stringset_bytesarray='wNZ7o'), Row(stringset_bytesarray='OKNV2'),Row(stringset_bytesarray='F0uC4'),Row(stringset_bytesarray='VBcyJ'),Row(stringset_bytesarray='KMZnH'))",
					"Count(Row(idset_long=839))",
					"Count(Row(id_long=809))",
					"Count(Row(idtq_long=533))",
					"Count(Row(idtq_long=533, from=\"2020-01-01\"))",
					"Count(Row(idset_int=533))",
					"Count(Row(id_int=168))",
					"Count(Row(idsettq_int=113))",
					"Row(idsettq_int=113, to=\"2024-01-01\")",
					"Count(Intersect(Row(idset_longarray=399),Row(idset_longarray=322), Row(idset_longarray=975), Row(idset_longarray=730), Row(idset_longarray=969)))",
					"Count(Intersect(Row(idtq_longarray=172),Row(idtq_longarray=388), Row(idtq_longarray=731), Row(idtq_longarray=429), Row(idtq_longarray=730)))",
					"Count(Intersect(Row(idtq_longarray=172, from=\"2022-01-01\"),Row(idtq_longarray=388, from=\"2022-01-01\"), Row(idtq_longarray=731, from=\"2022-01-01\"), Row(idtq_longarray=429, from=\"2022-01-01\"), Row(idtq_longarray=730, from=\"2022-01-01\")))",
					"Count(Intersect(Row(idset_intarray=958),Row(idset_intarray=242), Row(idset_intarray=778), Row(idset_intarray=289), Row(idset_intarray=797)))",
					"Count(Row(int_long > 500))",
					"Count(Row(int_int > 500))",
					"Count(Row(decimal_bytes > 1000.00))",
					"Count(Row(decimal_float > 3.05))",
					"Count(Row(decimal_double > 4.11))",
					"Count(Row(dateint_bytes_ts > 1675163490))",
					"Count(Row(bools=bool_bool))",
					"Count(Not(Row(bools=bool_bool)))",
					"Count(Row(timestamp_bytes_ts > \"2023-02-20T00:00:00Z\"))",
					"Count(Row(timestamp_bytes_int > \"2023-02-20T00:00:00Z\"))",
				},
			},
			expectedResults: [][]string{
				{
					"{\"results\":[10]}\n",
					"{\"results\":[2]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[0]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[2]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"DY2Ui|kUbdU|pjxqm\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n",
					"{\"results\":[{\"columns\":[]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"u2Yr4|sHaUv|x5z8P\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"yg8hY|tvNOB|byHh9\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"yg8hY|tvNOB|byHh9\"]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"6TKzc|YKLk9|h1iqc\"]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[4]}\n",
					"{\"results\":[5]}\n",
					"{\"results\":[4]}\n",
					"{\"results\":[3]}\n",
					"{\"results\":[3]}\n",
					"{\"results\":[8]}\n",
					"{\"results\":[4]}\n",
					"{\"results\":[6]}\n",
					"{\"results\":[3]}\n",
					"{\"results\":[3]}\n",
				},
			},
		},
		{ // confirm behavior of nulls.. when all null values passed, nothing should change
			name:              "all values can be inserted",
			pathsToAvroSchema: []string{"alltypes.json"},
			pathsToRecords:    []string{"./testdata/records/alltypes_null.json"},
			pilosaHosts:       pilosaHost,
			registryURL:       registryHost,
			kafkaHost:         kafkaHost,
			consumerConfigs: []ConsumerTestConfig{
				{
					idType:    "string",
					keyFields: []string{"pk0", "pk1", "pk2"},
					topic:     "alltypes_null",
					delete:    false,
				},
			},
			index: "alltypes_null",
			queries: [][]string{
				{
					//"Extract(All(), Rows(stringset_string), Rows(string_string), Rows(stringtq_string), Rows(stringset_bytes), Rows(string_bytes), Rows(stringsettq_bytes), Rows(stringset_stringarray), Rows(stringtq_stringarray), Rows(stringset_bytesarray), Rows(stringtq_bytesarray), Rows(idset_long), Rows(id_long), Rows(idtq_long), Rows(idset_int), Rows(id_int), Rows(idsettq_int), Rows(idset_longarray), Rows(idtq_longarray), Rows(idset_intarray), Rows(stringtq_stringarray), Rows(stringset_bytesarray), Rows(stringtq_bytesarray), Rows(idset_long), Rows(id_long), Rows(idtq_long), Rows(idset_int), Rows(id_int), Rows(idsettq_int), Rows(idset_longarray), Rows(idtq_longarray), Rows(idset_intarray), Rows(int_long), Rows(int_int), Rows(decimal_bytes), Rows(decimal_float), Rows(decimal_double), Rows(dateint_bytes_ts), Rows(bools), Rows(timestamp_bytes_ts), Rows(timestamp_bytes_int))",
					"Count(All())",
					"Count(Row(stringset_string='7EYSp'))",
					"Count(Row(string_string='uirDR'))",
					"Count(Row(stringtq_string='Qylqq'))",
					"Count(Row(stringset_bytes='gL2Hg'))",
					"Count(Row(string_bytes='BmvHF'))",
					"Count(Row(stringsettq_bytes='798ka'))",
					"Count(Intersect(Row(stringset_stringarray='vbbuf'), Row(stringset_stringarray='VQs7y'), Row(stringset_stringarray='9z4aw'), Row(stringset_stringarray='h1iqc'), Row(stringset_stringarray='aQQxr')))",
					"Count(Intersect(Row(stringtq_stringarray='x5z8P'), Row(stringtq_stringarray='0UGJQ'), Row(stringtq_stringarray='58KIR'), Row(stringtq_stringarray='7EYSp'), Row(stringtq_stringarray='CKs1F')))",
					"Count(Intersect(Row(stringset_bytesarray='u2Yr4'), Row(stringset_bytesarray='tvNOB'), Row(stringset_bytesarray='iYeOV'), Row(stringset_bytesarray='ZgkOB'), Row(stringset_bytesarray='RPGAm')))",
					"Count(Intersect(Row(stringtq_bytesarray='BwqU2'), Row(stringtq_bytesarray='6iGIm'), Row(stringtq_bytesarray='fjQK2'), Row(stringtq_bytesarray='LBTEU'), Row(stringtq_bytesarray='C6xxn')))",
					"Count(Row(idset_long=647))",
					"Count(Row(id_long=792))",
					"Count(Row(idtq_long=676))",
					"Count(Row(idset_int=898))",
					"Count(Row(id_int=63))",
					"Count(Row(idsettq_int=890))",
					"Count(Intersect(Row(idset_longarray=442), Row(idset_longarray=167), Row(idset_longarray=230), Row(idset_longarray=344), Row(idset_longarray=733)))",
					"Count(Intersect(Row(idtq_longarray=385), Row(idtq_longarray=931), Row(idtq_longarray=157), Row(idtq_longarray=865), Row(idtq_longarray=394)))",
					"Count(Intersect(Row(idset_intarray=442), Row(idset_intarray=614), Row(idset_intarray=394), Row(idset_intarray=284), Row(idset_intarray=344)))",
					"Count(Row(int_long=584))",
					"Count(Row(int_int=344))",
					"Count(Row(decimal_bytes=1155.95))",
					"Count(Row(decimal_float=3.23))",
					"Count(Row(decimal_double=0.95))",
					"Count(Row(dateint_bytes_ts=1676534039))",
					"Count(Row(bools=bool_bool))",
					"Count(Row(timestamp_bytes_ts='2023-02-16T07:53:59Z'))",
					"Count(Row(timestamp_bytes_int=1676555639))",
				},
			},
			expectedResults: [][]string{
				{
					//"{\"results\":[{\"fields\":[{\"name\":\"stringset_string\",\"type\":\"[]string\"},{\"name\":\"string_string\",\"type\":\"string\"},{\"name\":\"stringtq_string\",\"type\":\"[]string\"},{\"name\":\"stringset_bytes\",\"type\":\"[]string\"},{\"name\":\"string_bytes\",\"type\":\"string\"},{\"name\":\"stringsettq_bytes\",\"type\":\"[]string\"},{\"name\":\"stringset_stringarray\",\"type\":\"[]string\"},{\"name\":\"stringtq_stringarray\",\"type\":\"[]string\"},{\"name\":\"stringset_bytesarray\",\"type\":\"[]string\"},{\"name\":\"stringtq_bytesarray\",\"type\":\"[]string\"},{\"name\":\"idset_long\",\"type\":\"[]uint64\"},{\"name\":\"id_long\",\"type\":\"uint64\"},{\"name\":\"idtq_long\",\"type\":\"[]uint64\"},{\"name\":\"idset_int\",\"type\":\"[]uint64\"},{\"name\":\"id_int\",\"type\":\"uint64\"},{\"name\":\"idsettq_int\",\"type\":\"[]uint64\"},{\"name\":\"idset_longarray\",\"type\":\"[]uint64\"},{\"name\":\"idtq_longarray\",\"type\":\"[]uint64\"},{\"name\":\"idset_intarray\",\"type\":\"[]uint64\"},{\"name\":\"stringtq_stringarray\",\"type\":\"[]string\"},{\"name\":\"stringset_bytesarray\",\"type\":\"[]string\"},{\"name\":\"stringtq_bytesarray\",\"type\":\"[]string\"},{\"name\":\"idset_long\",\"type\":\"[]uint64\"},{\"name\":\"id_long\",\"type\":\"uint64\"},{\"name\":\"idtq_long\",\"type\":\"[]uint64\"},{\"name\":\"idset_int\",\"type\":\"[]uint64\"},{\"name\":\"id_int\",\"type\":\"uint64\"},{\"name\":\"idsettq_int\",\"type\":\"[]uint64\"},{\"name\":\"idset_longarray\",\"type\":\"[]uint64\"},{\"name\":\"idtq_longarray\",\"type\":\"[]uint64\"},{\"name\":\"idset_intarray\",\"type\":\"[]uint64\"}],\"columns\":[{\"column\":\"9z4aw|5ptDx|CKs1F\",\"rows\":[[\"7EYSp\"],\"uirDR\",[\"Qylqq\"],[\"gL2Hg\"],\"BmvHF\",[\"798ka\"],[\"9z4aw\",\"h1iqc\",\"aQQxr\",\"vbbuf\",\"VQs7y\"],[\"7EYSp\",\"CKs1F\",\"x5z8P\",\"0UGJQ\",\"58KIR\"],[\"u2Yr4\",\"tvNOB\",\"iYeOV\",\"ZgkOB\",\"RPGAm\"],[\"BwqU2\",\"6iGIm\",\"fjQK2\",\"LBTEU\",\"C6xxn\"],[647],792,[676],[898],63,[890],[167,230,344,442,733],[157,385,394,865,931],[284,344,394,442,614],[\"7EYSp\",\"CKs1F\",\"x5z8P\",\"0UGJQ\",\"58KIR\"],[\"u2Yr4\",\"tvNOB\",\"iYeOV\",\"ZgkOB\",\"RPGAm\"],[\"BwqU2\",\"6iGIm\",\"fjQK2\",\"LBTEU\",\"C6xxn\"],[647],792,[676],[898],63,[890],[167,230,344,442,733],[157,385,394,865,931],[284,344,394,442,614]]}]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
				},
			},
		},
		{ // confirm delete behavior when index is keyed
			name: "all values can be deleted with keys",
			pathsToAvroSchema: []string{
				"alltypes.json",
				"alltypes_delete_fields.json",
				"alltypes_delete_records.json",
				"alltypes_delete_value.json",
			},
			pathsToRecords: []string{
				"./testdata/records/alltypes.json",
				"./testdata/records/alltypes_delete_fields.json",
				"./testdata/records/alltypes_delete_records.json",
				"./testdata/records/alltypes_delete_value.json",
			},
			pilosaHosts: pilosaHost,
			registryURL: registryHost,
			kafkaHost:   kafkaHost,
			consumerConfigs: []ConsumerTestConfig{
				{
					idType:    "string",
					keyFields: []string{"pk0", "pk1", "pk2"},
					topic:     "alltypes",
					delete:    false,
				},
				{
					idType:    "string",
					keyFields: []string{"pk0", "pk1", "pk2"},
					topic:     "alltypes_delete_fields",
					delete:    true,
				},
				{
					idType:    "string",
					keyFields: []string{"pk0", "pk1", "pk2"},
					topic:     "alltypes_delete_record",
					delete:    true,
				},
				{
					idType:    "string",
					keyFields: []string{"pk0", "pk1", "pk2"},
					topic:     "alltypes_delete_values",
					delete:    true,
				},
			},
			index: "alltypes_delete",
			queries: [][]string{
				{
					"Count(All())",
					"Count(ConstRow(columns=['u2Yr4|sHaUv|x5z8P', 'DY2Ui|kUbdU|pjxqm']))",
					"Count(Row(stringset_string='58KIR'))",
					"Count(Row(string_string='8MGwy'))",
					"Count(Row(stringtq_string='ivWWb'))",
					"Count(Row(stringtq_string='ivWWb', to=\"2023-02-03\"))",
					"Count(Row(stringtq_string='ivWWb', from=\"2023-02-03\", to=\"2023-02-04\"))",
					"Count(Union(Row(stringset_bytes='eNKWF'),Row(stringset_bytes='5ptDx')))",
					"Row(string_bytes='vTwn4')",
					"Row(stringsettq_bytes='798ka')",
					"Row(stringsettq_bytes='798ka', from=\"2023-02-18\")",
					"Row(stringsettq_bytes='798ka', from=\"2023-02-16\", to=\"2023-02-18\")",
					"Intersect(Row(stringset_stringarray='u2Yr4'), Row(stringset_stringarray='PYE8V'), Row(stringset_stringarray='VBcyJ'), Row(stringset_stringarray='Chgzr'), Row(stringset_stringarray='DY2Ui'))",
					"Row(stringtq_stringarray='oxjI0', from=\"2023-01-29\", to=\"2023-01-31\")",
					"Intersect(Row(stringset_bytesarray='wNZ7o'), Row(stringset_bytesarray='OKNV2'),Row(stringset_bytesarray='F0uC4'),Row(stringset_bytesarray='VBcyJ'),Row(stringset_bytesarray='KMZnH'))",
					"Count(Row(idset_long=839))",
					"Count(Row(id_long=809))",
					"Count(Row(idtq_long=533))",
					"Count(Row(idtq_long=533, from=\"2020-01-01\"))",
					"Count(Row(idset_int=533))",
					"Count(Row(id_int=168))",
					"Count(Row(idsettq_int=113))",
					"Row(idsettq_int=113, to=\"2024-01-01\")",
					"Count(Intersect(Row(idset_longarray=399),Row(idset_longarray=322), Row(idset_longarray=975), Row(idset_longarray=730), Row(idset_longarray=969)))",
					"Count(Intersect(Row(idtq_longarray=172),Row(idtq_longarray=388), Row(idtq_longarray=731), Row(idtq_longarray=429), Row(idtq_longarray=730)))",
					"Count(Intersect(Row(idtq_longarray=172, from=\"2022-01-01\"),Row(idtq_longarray=388, from=\"2022-01-01\"), Row(idtq_longarray=731, from=\"2022-01-01\"), Row(idtq_longarray=429, from=\"2022-01-01\"), Row(idtq_longarray=730, from=\"2022-01-01\")))",
					"Count(Intersect(Row(idset_intarray=958),Row(idset_intarray=242), Row(idset_intarray=778), Row(idset_intarray=289), Row(idset_intarray=797)))",
					"Count(Row(int_long > 500))",
					"Count(Row(int_int > 500))",
					"Count(Row(decimal_bytes > 1000.00))",
					"Count(Row(decimal_float > 3.05))",
					"Count(Row(decimal_double > 4.11))",
					"Count(Row(dateint_bytes_ts > 1675163490))",
					"Count(Row(bools=bool_bool))",
					"Count(Not(Row(bools=bool_bool)))",
					"Count(Row(timestamp_bytes_ts > \"2023-02-20T00:00:00Z\"))",
					"Count(Row(timestamp_bytes_int > \"2023-02-20T00:00:00Z\"))",
				},
				{
					"Count(All())",
					"Row(int_long=null)",
					"Count(UnionRows(Rows(stringset_stringarray)))",
					"Not(UnionRows(Rows(stringset_stringarray), Rows(string_string),Rows(stringset_bytes),Rows(string_bytes),Rows(stringset_stringarray),Rows(stringset_bytesarray),Rows(idset_long),Rows(id_long),Rows(idset_int),Rows(id_int),Rows(idset_longarray),Rows(idset_intarray)))",
					"Row(int_int=null)",
					"Row(decimal_bytes=null)",
					"Row(decimal_float=null)",
					"Count(Row(bools=bool_bool))",
					"Count(Not(Row(bools-exists=bool_bool)))",
					"Count(Row(dateint_bytes_ts=null))",
				},
				{
					"Count(All())",
				},
				{
					"Row(string_string=\"ZgkOB\")",
					"Row(stringset_string =\"7EYSp\")",
					"Count(Not(UnionRows(Rows(stringset_string))))",
					"Intersect(Not(Intersect(Row(stringset_stringarray=\"u2Yr4\"),Row(stringset_stringarray=\"PYE8V\"), Row(stringset_stringarray=\"VBcyJ\"))), Intersect(Row(stringset_stringarray=\"Chgzr\"), Row(stringset_stringarray=\"DY2Ui\")))",
					"Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Not(UnionRows(Rows(idset_int)))))",
					"Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Not(UnionRows(Rows(id_int)))))",
					"Row(int_int=969)",
					"Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Not(UnionRows(Rows(bools-exists)))))",
					"Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Row(decimal_double=null)))",
					"Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Not(Row(dateint_bytes_ts=null))))",
					"Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Row(timestamp_bytes_int=null)))",
				},
			},
			expectedResults: [][]string{
				{
					"{\"results\":[10]}\n",
					"{\"results\":[2]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[0]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[2]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"DY2Ui|kUbdU|pjxqm\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n",
					"{\"results\":[{\"columns\":[]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"u2Yr4|sHaUv|x5z8P\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"yg8hY|tvNOB|byHh9\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"yg8hY|tvNOB|byHh9\"]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"6TKzc|YKLk9|h1iqc\"]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[4]}\n",
					"{\"results\":[5]}\n",
					"{\"results\":[4]}\n",
					"{\"results\":[3]}\n",
					"{\"results\":[3]}\n",
					"{\"results\":[8]}\n",
					"{\"results\":[4]}\n",
					"{\"results\":[6]}\n",
					"{\"results\":[3]}\n",
					"{\"results\":[3]}\n",
				},
				{
					"{\"results\":[10]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n",
					"{\"results\":[9]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"6TKzc|YKLk9|h1iqc\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"RKE3c|6TKzc|RKE3c\"]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"RKE3c|6TKzc|RKE3c\"]}]}\n",
					"{\"results\":[3]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
				},
				{
					"{\"results\":[7]}\n",
				},
				{
					"{\"results\":[{\"columns\":[]}]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"u2Yr4|sHaUv|x5z8P\"]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"u2Yr4|sHaUv|x5z8P\"]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[{\"columns\":[],\"keys\":[\"u2Yr4|sHaUv|x5z8P\"]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
				},
			},
		},
		{ // confirm delete behavior when index is not keyed (note field level delete not supported with delete consumer on keys: false index)
			name: "all values can be deleted with id keys",
			pathsToAvroSchema: []string{
				"alltypes.json",
				"alltypes_delete_records.json",
				"alltypes_delete_value.json",
			},
			pathsToRecords: []string{
				"./testdata/records/alltypes.json",
				"./testdata/records/alltypes_delete_records_ids.json",
				"./testdata/records/alltypes_delete_value_ids.json",
			},
			pilosaHosts: pilosaHost,
			registryURL: registryHost,
			kafkaHost:   kafkaHost,
			consumerConfigs: []ConsumerTestConfig{
				{
					idType: "generate",
					topic:  "alltypes",
					delete: false,
				},
				{
					idType: "generate",
					topic:  "alltypes_delete_record_ids",
					delete: true,
				},
				{
					idType: "generate",
					topic:  "alltypes_delete_values_ids",
					delete: true,
				},
			},
			index: "alltypes_delete_ids",
			queries: [][]string{
				{
					"Count(All())",
					"Count(ConstRow(columns=[2, 3]))",
					"Count(Row(stringset_string='58KIR'))",
					"Count(Row(string_string='8MGwy'))",
					"Count(Row(stringtq_string='ivWWb'))",
					"Count(Row(stringtq_string='ivWWb', to=\"2023-02-03\"))",
					"Count(Row(stringtq_string='ivWWb', from=\"2023-02-03\", to=\"2023-02-04\"))",
					"Count(Union(Row(stringset_bytes='eNKWF'),Row(stringset_bytes='5ptDx')))",
					"Count(Row(string_bytes='vTwn4'))",
					"Count(Row(stringsettq_bytes='798ka'))",
					"Count(Row(stringsettq_bytes='798ka', from=\"2023-02-18\"))",
					"Count(Row(stringsettq_bytes='798ka', from=\"2023-02-16\", to=\"2023-02-18\"))",
					"Intersect(Row(stringset_stringarray='u2Yr4'), Row(stringset_stringarray='PYE8V'), Row(stringset_stringarray='VBcyJ'), Row(stringset_stringarray='Chgzr'), Row(stringset_stringarray='DY2Ui'))",
					"Row(stringtq_stringarray='oxjI0', from=\"2023-01-29\", to=\"2023-01-31\")",
					"Intersect(Row(stringset_bytesarray='wNZ7o'), Row(stringset_bytesarray='OKNV2'),Row(stringset_bytesarray='F0uC4'),Row(stringset_bytesarray='VBcyJ'),Row(stringset_bytesarray='KMZnH'))",
					"Count(Row(idset_long=839))",
					"Count(Row(id_long=809))",
					"Count(Row(idtq_long=533))",
					"Count(Row(idtq_long=533, from=\"2020-01-01\"))",
					"Count(Row(idset_int=533))",
					"Count(Row(id_int=168))",
					"Count(Row(idsettq_int=113))",
					"Row(idsettq_int=113, to=\"2024-01-01\")",
					"Count(Intersect(Row(idset_longarray=399),Row(idset_longarray=322), Row(idset_longarray=975), Row(idset_longarray=730), Row(idset_longarray=969)))",
					"Count(Intersect(Row(idtq_longarray=172),Row(idtq_longarray=388), Row(idtq_longarray=731), Row(idtq_longarray=429), Row(idtq_longarray=730)))",
					"Count(Intersect(Row(idtq_longarray=172, from=\"2022-01-01\"),Row(idtq_longarray=388, from=\"2022-01-01\"), Row(idtq_longarray=731, from=\"2022-01-01\"), Row(idtq_longarray=429, from=\"2022-01-01\"), Row(idtq_longarray=730, from=\"2022-01-01\")))",
					"Count(Intersect(Row(idset_intarray=958),Row(idset_intarray=242), Row(idset_intarray=778), Row(idset_intarray=289), Row(idset_intarray=797)))",
					"Count(Row(int_long > 500))",
					"Count(Row(int_int > 500))",
					"Count(Row(decimal_bytes > 1000.00))",
					"Count(Row(decimal_float > 3.05))",
					"Count(Row(decimal_double > 4.11))",
					"Count(Row(dateint_bytes_ts > 1675163490))",
					"Count(Row(bools=bool_bool))",
					"Count(Not(Row(bools=bool_bool)))",
					"Count(Row(timestamp_bytes_ts > \"2023-02-20T00:00:00Z\"))",
					"Count(Row(timestamp_bytes_int > \"2023-02-20T00:00:00Z\"))",
				},
				{
					"Count(All())",
				},
				{
					"Row(string_string=\"ZgkOB\")",
					"Row(stringset_string =\"7EYSp\")",
					"Intersect(Not(Intersect(Row(stringset_stringarray=\"u2Yr4\"),Row(stringset_stringarray=\"PYE8V\"), Row(stringset_stringarray=\"VBcyJ\"))), Intersect(Row(stringset_stringarray=\"Chgzr\"), Row(stringset_stringarray=\"DY2Ui\")))",
					"Count(Intersect(ConstRow(columns=[10]), Not(UnionRows(Rows(idset_int)))))",
					"Count(Intersect(ConstRow(columns=[10]), Not(UnionRows(Rows(id_int)))))",
					"Row(int_int=969)",
					"Row(int_int=null)",
					"Count(Intersect(ConstRow(columns=[10]), Not(UnionRows(Rows(bools-exists)))))",
					"Count(Intersect(ConstRow(columns=[10]), Row(decimal_double=null)))",
					"Count(Intersect(ConstRow(columns=[10]), Not(Row(dateint_bytes_ts=null))))",
					"Count(Intersect(ConstRow(columns=[10]), Row(timestamp_bytes_int=null)))",
				},
			},
			expectedResults: [][]string{
				{
					"{\"results\":[10]}\n",
					"{\"results\":[2]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[0]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[2]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[0]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[{\"columns\":[10]}]}\n",
					"{\"results\":[{\"columns\":[6]}]}\n",
					"{\"results\":[{\"columns\":[6]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[{\"columns\":[7]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[4]}\n",
					"{\"results\":[5]}\n",
					"{\"results\":[4]}\n",
					"{\"results\":[3]}\n",
					"{\"results\":[3]}\n",
					"{\"results\":[8]}\n",
					"{\"results\":[4]}\n",
					"{\"results\":[6]}\n",
					"{\"results\":[3]}\n",
					"{\"results\":[3]}\n",
				},
				{
					"{\"results\":[7]}\n",
				},
				{
					"{\"results\":[{\"columns\":[]}]}\n",
					"{\"results\":[{\"columns\":[]}]}\n",
					"{\"results\":[{\"columns\":[10]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[{\"columns\":[10]}]}\n",
					"{\"results\":[{\"columns\":[8]}]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
					"{\"results\":[1]}\n",
				},
			},
		},
	}

	for _, test := range tests {

		// define some vars
		now := time.Now().UnixNano()
		index := fmt.Sprintf("%s_%d", test.index, now)
		//index := test.index

		// so topics of the same name end up being the same name after
		// adding a timestamp to make sure they're unique
		var topics map[string]string
		for _, iteration := range test.consumerConfigs {
			if topicName, ok := topics[iteration.topic]; ok {
				iteration.topic = topicName
			} else {
				newTopicName := fmt.Sprintf("%s_%d", iteration.topic, now)
				iteration.topic = newTopicName

			}
		}

		// iterate through tests (one per consumer config)
		for i, iteration := range test.consumerConfigs {
			// read in records
			records, err := kafkaRecordFromFile(test.pathsToRecords[i])
			if err != nil {
				t.Errorf("%s", err)
			}

			// load schema registry, create and produce to the topic
			writeRecordsToKafka(t, test.pathsToAvroSchema[i], iteration.topic, test.registryURL, test.kafkaHost, records)

			// configure the consumer
			consumer, err := NewMain()
			if err != nil {
				t.Fatalf("creating main %v", err)
			}
			configureTestFlags(consumer)
			consumer.Index = index
			consumer.Topics = []string{iteration.topic}
			consumer.KafkaBootstrapServers = []string{test.kafkaHost}
			consumer.SchemaRegistryURL = test.registryURL
			consumer.Delete = iteration.delete
			switch iteration.idType {
			case "id":
				consumer.IDField = iteration.keyFields[0]
			case "string":
				consumer.PrimaryKeyFields = iteration.keyFields
			case "generate":
				consumer.AutoGenerate = true
				consumer.ExternalGenerate = true
			default:
				t.Errorf("incorrect idType supplied")
			}
			consumer.MaxMsgs = uint64(len(records))
			consumer.BatchSize = int(consumer.MaxMsgs)
			pilosaHostsNew := strings.Split(test.pilosaHosts, ",")
			consumer.PilosaHosts = pilosaHostsNew
			//remove after flipping back to docker
			host := strings.Split(pilosaHostsNew[0], ":")
			consumer.PilosaGRPCHosts = []string{fmt.Sprintf("%s:20101", host[0])}

			// consumer records
			err = consumer.Run()
			if err != nil {
				t.Fatalf("running consumer: %v", err)
			}

			// now run queries and confirm the data is as expected
			client := consumer.PilosaClient()
			runTestQueries(t, index, test.queries[i], test.expectedResults[i], client)
		}
	}
}

func runTestQueries(t *testing.T, index string, queries, expectedResults []string, client *pilosaclient.Client) {
	for i, q := range queries {
		status, body, err := client.HTTPRequest("POST", fmt.Sprintf("/index/%s/query", index), []byte(q), nil)
		if err != nil {
			t.Fatalf("querying featurebase: status: %d, response: %s, error: %s: on query: %s", status, body, err, q)
		}
		if string(body[:]) != expectedResults[i] {
			t.Fatalf("running query %s against index %s. expected %s but got %s", q, index, expectedResults[i], body)
		}
	}
}

func kafkaRecordFromFile(pathToRecords string) (records []map[string]interface{}, err error) {
	var data map[string]interface{}

	recordsFile, err := os.Open(pathToRecords)
	if err != nil {
		return nil, fmt.Errorf("opening records file")
	}
	defer recordsFile.Close()

	s := bufio.NewScanner(recordsFile)
	for s.Scan() {
		err := json.Unmarshal(s.Bytes(), &data)
		if err != nil {
			return nil, fmt.Errorf("unmarshal json: %s", err)
		}
		records = append(records, data)
		data = make(map[string]interface{})
	}
	return records, nil
}

func writeRecordsToKafka(t *testing.T, pathToAvroSchema, topic, schemaRegistryURL, kafkaHost string, records []map[string]interface{}) {
	licodec := liDecodeTestSchema(t, pathToAvroSchema)
	schemaID := postSchema(t, pathToAvroSchema, fmt.Sprintf("%s_id", topic), schemaRegistryURL, nil)
	p, err := confluent.NewProducer(&confluent.ConfigMap{
		"bootstrap.servers": kafkaHost,
	})
	if err != nil {
		t.Fatalf("Failed to create producer: %s", err)
	}
	defer p.Close()
	tCreateTopic(t, topic, p)

	tPutRecordsKafka(t, p, topic, schemaID, licodec, "akey", records...)
}

type sortableCRI []pilosaclient.CountResultItem

func (s sortableCRI) Len() int { return len(s) }
func (s sortableCRI) Less(i, j int) bool {
	if s[i].Count != s[j].Count {
		return s[i].Count > s[j].Count
	}
	if s[i].ID != s[j].ID {
		return s[i].ID < s[j].ID
	}
	if s[i].Key != s[j].Key {
		return s[i].Key < s[j].Key
	}
	return true
}

func (s sortableCRI) Swap(i, j int) {
	s[i], s[j] = s[j], s[i]
}

func makeRecord(t *testing.T, fields []string, vals []interface{}) map[string]interface{} {
	if len(fields) != len(vals) {
		t.Fatalf("have %d fields and %d vals", len(fields), len(vals))
	}
	ret := make(map[string]interface{})
	for i, field := range fields {
		ret[field] = vals[i]
	}
	return ret
}

func tDoHTTPPost(t *testing.T, url, contentType, body string) string {
	resp, err := http.Post(url, contentType, strings.NewReader(body))
	if err != nil {
		t.Fatalf("making POST request: %v", err)
	}

	bod, err := io.ReadAll(resp.Body)
	if err != nil {
		t.Fatalf("reading POST response bdoy: %v", err)
	}

	return string(bod)
}
