scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-12 15:51:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

FB-1618: Fix code smells due to unifying IDK and Featurebase (#2184)

Browse source 
* create constants for duplicates to resolve code smells

* refactored duplicated code for extract(rows))

* additional refactoring

* refactor test

* rename to match reg exp

* fixed naming

* fixed naming, removed duplicated string

* fixed index names

* fixed bug introduced in earlier commit

* create delete index method and refactor if statement

* refactored file to reduce complexity

* fix go fmt error

* fixed bug introduced in last commit

* address reviewer's comments

* remove trailing colon and spaces
This commit is contained in:
souhailanoor 2022-08-12 17:00:29 -05:00 • committed by GitHub
parent 985a70b14e
commit 39def696c5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 474 additions and 547 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

306
idk/csv/csv_test.go
View file

@ -16,6 +16,8 @@ import (
"github.com/molecula/featurebase/v3/idk/idktest"
)
const TestName = "csvtest"
func configureTestFlags(main *Main) {
if pilosaHost, ok := os.LookupEnv("IDK_TEST_PILOSA_HOST"); ok {
main.PilosaHosts = []string{pilosaHost}
@ -30,6 +32,54 @@ func configureTestFlags(main *Main) {
main.Stats = ""
}
// helper method to run "Extract(All(), Rows())" queries and compare results
func testExtractRowsQuery(t *testing.T, index string, field string, check []interface{}) {
pql := fmt.Sprintf("Extract(All(), Rows(%s))", field)
eResp, err := idktest.DoExtractQuery(pql, index)
if err != nil {
t.Fatal("doing extract: ", err)
}
if eResp.Results[0].Columns == nil {
t.Fatal("no results: ", err)
}
for i, item := range eResp.Results[0].Columns {
switch exp := check[i].(type) {
case nil:
if item.Rows[0] != nil {
t.Errorf("expected nil, got %+v", item.Rows[0])
}
case string:
if item.Rows[0] != exp {
t.Errorf("expected %s, got %+v", exp, item.Rows[0])
}
case int:
vAsInt := int(item.Rows[0].(float64))
if vAsInt != exp {
t.Errorf("expected %d, got %+v", exp, item.Rows[0])
} else {
expAsFloat := float64(exp)
if item.Rows[0] != expAsFloat {
t.Errorf("expected %f, got %+v", expAsFloat, item.Rows[0])
}
}
case float64:
if item.Rows[0] != exp {
t.Errorf("expected %f, got %+v", exp, item.Rows[0])
}
default:
t.Errorf("unknown type: %T", exp)
}
}
}
func testDeleteIndex(t *testing.T, m *Main) {
if err := m.PilosaClient().DeleteIndexByName(m.Index); err != nil {
t.Logf(idktest.ErrDeletingIndex+"%v", err)
}
}
func TestCSVCommand(t *testing.T) {
file := `
id__ID,s__String_F_YMDH,__RecordTime_2006-01-02T15
@ -47,24 +97,19 @@ id__ID,s__String_F_YMDH,__RecordTime_2006-01-02T15
configureTestFlags(m)
m.Files = []string{name}
rand.Seed(time.Now().UnixNano())
m.Index = fmt.Sprintf("csvtest%d", rand.Intn(100000))
defer func() {
if err := m.PilosaClient().DeleteIndexByName(m.Index); err != nil {
t.Logf("deleting test index: %v", err)
}
}()
m.Index = fmt.Sprintf(TestName+"%d", rand.Intn(100000))
defer testDeleteIndex(t, m)
err := m.Run()
if err != nil {
t.Fatalf("running: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
client := m.PilosaClient()
schema, err := client.Schema()
if err != nil {
t.Fatalf("getting schema: %v", err)
t.Fatalf("%s: %v", idktest.ErrGettingSchema, err)
}
index := schema.Index(m.Index)
@ -72,7 +117,7 @@ id__ID,s__String_F_YMDH,__RecordTime_2006-01-02T15
resp, err := client.Query(s.Range("a", tim(t, "2019-01-07T03"), tim(t, "2019-01-10T05")))
if err != nil {
t.Fatalf("querying: %v", err)
t.Fatalf("%s: %v", idktest.ErrGettingQuery, err)
}
if !reflect.DeepEqual(resp.Results()[0].Row().Columns, []uint64{0, 1, 5}) {
@ -99,24 +144,19 @@ ABCD,2019-01-30,40%
configureTestFlags(m)
m.Files = []string{name}
rand.Seed(time.Now().UnixNano())
m.Index = fmt.Sprintf("csvtest%d", rand.Intn(100000))
defer func() {
if err := m.PilosaClient().DeleteIndexByName(m.Index); err != nil {
t.Logf("deleting test index: %v", err)
}
}()
m.Index = fmt.Sprintf(TestName+"%d", rand.Intn(100000))
defer testDeleteIndex(t, m)
err := m.Run()
if err != nil {
t.Fatalf("running: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
client := m.PilosaClient()
schema, err := client.Schema()
if err != nil {
t.Fatalf("getting schema: %v", err)
t.Fatalf("%s: %v", idktest.ErrGettingSchema, err)
}
index := schema.Index(m.Index)
@ -124,7 +164,7 @@ ABCD,2019-01-30,40%
resp, err := client.Query(s.Row("ABCD"))
if err != nil {
t.Fatalf("querying: %v", err)
t.Fatalf("%s: %v", idktest.ErrGettingQuery, err)
}
if !reflect.DeepEqual(resp.Results()[0].Row().Columns, []uint64{0, 1, 2, 7}) {
@ -176,24 +216,19 @@ func TestCSVRecordTime(t *testing.T) {
configureTestFlags(m)
m.Files = []string{name}
rand.Seed(time.Now().UnixNano())
m.Index = fmt.Sprintf("csvtest%d", rand.Intn(100000))
defer func() {
if err := m.PilosaClient().DeleteIndexByName(m.Index); err != nil {
t.Logf("deleting test index: %v", err)
}
}()
m.Index = fmt.Sprintf(TestName+"%d", rand.Intn(100000))
defer testDeleteIndex(t, m)
err := m.Run()
if err != nil {
t.Fatalf("running: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
client := m.PilosaClient()
schema, err := client.Schema()
if err != nil {
t.Fatalf("getting schema: %v", err)
t.Fatalf("%s: %v", idktest.ErrGettingSchema, err)
}
index := schema.Index(m.Index)
@ -204,7 +239,7 @@ func TestCSVRecordTime(t *testing.T) {
// not convinced that this is correct.
resp, err := client.Query(s.Range("a", tim(t, "2019-01-09T03"), tim(t, "2019-01-09T05").Add(1*time.Minute)))
if err != nil {
t.Fatalf("querying: %v", err)
t.Fatalf("%s: %v", idktest.ErrGettingQuery, err)
}
if !reflect.DeepEqual(resp.Results()[0].Row().Columns, []uint64{1, 2, 3}) {
@ -328,7 +363,8 @@ id__ID,s__String_F_YMDH,ts__Timestamp_s_2006-01-02 15:04:05.999,price__Decimal_2
name := writeTempFile(t, file)
checker := make(map[string]interface{})
checker["ts"] = []interface{}{nil, nil, nil, nil, nil, nil, nil, "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", "1500-04-03T00:00:00Z", "2019-04-03T00:00:00Z"}
ts := "2019-04-03T00:00:00Z"
checker["ts"] = []interface{}{nil, nil, nil, nil, nil, nil, nil, ts, ts, ts, ts, ts, "1500-04-03T00:00:00Z", ts}
checker["age"] = []interface{}{1, 35, 120, 120, 120, nil, 120, 1, 1, 100, nil, nil, nil, 100}
checker["price"] = []interface{}{0.0, 5.44, 5.44, 5.44, 5.44, 5.44, 5.44, 123.12, -1.0, nil, 2.34, 3.44, nil, 3.44}
@ -338,81 +374,37 @@ id__ID,s__String_F_YMDH,ts__Timestamp_s_2006-01-02 15:04:05.999,price__Decimal_2
configureTestFlags(m)
m.Files = []string{name}
rand.Seed(time.Now().UnixNano())
m.Index = fmt.Sprintf("csvtest%d", rand.Intn(100000))
m.Index = fmt.Sprintf(TestName+"%d", rand.Intn(100000))
m.AllowIntOutOfRange = true
m.AllowDecimalOutOfRange = true
m.AllowTimestampOutOfRange = true
defer func() {
if err := m.PilosaClient().DeleteIndexByName(m.Index); err != nil {
t.Logf("deleting test index: %v", err)
}
}()
defer testDeleteIndex(t, m)
err := m.Run()
if err != nil {
t.Fatalf("running: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
client := m.PilosaClient()
schema, err := client.Schema()
if err != nil {
t.Fatalf("getting schema: %v", err)
t.Fatalf("%s: %v", idktest.ErrGettingSchema, err)
}
index := schema.Index(m.Index)
resp, err := client.Query(index.Count(index.All()))
if err != nil {
t.Fatalf("querying: %v", err)
t.Fatalf("%s: %v", idktest.ErrGettingQuery, err)
}
if cnt := resp.Result().Count(); cnt != 14 {
t.Fatalf("expected 14, got %+v", cnt)
}
for _, field := range []string{"ts", "age", "price"} {
pql := fmt.Sprintf("Extract(All(), Rows(%s))", field)
eResp, err := idktest.DoExtractQuery(pql, m.Index)
if err != nil {
t.Fatal("doing extract: ", err)
}
if eResp.Results[0].Columns == nil {
t.Fatal("no results: ", err)
}
for i, item := range eResp.Results[0].Columns {
check := checker[field].([]interface{})
switch exp := check[i].(type) {
case nil:
if item.Rows[0] != nil {
t.Errorf("expected nil, got %+v", item.Rows[0])
}
case string:
if item.Rows[0] != exp {
t.Errorf("expected %s, got %+v", exp, item.Rows[0])
}
case int:
vAsInt := int(item.Rows[0].(float64))
if vAsInt != exp {
t.Errorf("expected %d, got %+v", exp, item.Rows[0])
} else {
expAsFloat := float64(exp)
if item.Rows[0] != expAsFloat {
t.Errorf("expected %f, got %+v", expAsFloat, item.Rows[0])
}
}
case float64:
if item.Rows[0] != exp {
t.Errorf("expected %f, got %+v", exp, item.Rows[0])
}
default:
t.Errorf("unknown type: %T", exp)
}
}
check := checker[field].([]interface{})
testExtractRowsQuery(t, m.Index, field, check)
}
}
type TimestampTestCase struct {
@ -489,56 +481,15 @@ id__ID,s__String_F_YMDH,gran-conversion__Timestamp_ns_2006-01-02T15:04:05.999999
}
func testTimestampRunner(t *testing.T, m *Main, testCase TimestampTestCase) {
defer func() {
if err := m.PilosaClient().DeleteIndexByName(m.Index); err != nil {
t.Logf("deleting test index: %v", err)
}
}()
defer testDeleteIndex(t, m)
err := m.Run()
if err != nil {
t.Fatalf("running: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
pql := fmt.Sprintf("Extract(All(), Rows(%s))", testCase.fieldName)
eResp, err := idktest.DoExtractQuery(pql, m.Index)
if err != nil {
t.Fatal("doing extract: ", err)
}
if eResp.Results[0].Columns == nil {
t.Fatal("no results: ", err)
}
for i, item := range eResp.Results[0].Columns {
check := testCase.expect.([]interface{})
switch exp := check[i].(type) {
case nil:
if item.Rows[0] != nil {
t.Errorf("expected nil, got %+v", item.Rows[0])
}
case string:
if item.Rows[0] != exp {
t.Errorf("expected %s, got %+v", exp, item.Rows[0])
}
case int:
vAsInt := int(item.Rows[0].(float64))
if vAsInt != exp {
t.Errorf("expected %d, got %+v", exp, item.Rows[0])
} else {
expAsFloat := float64(exp)
if item.Rows[0] != expAsFloat {
t.Errorf("expected %f, got %+v", expAsFloat, item.Rows[0])
}
}
case float64:
if item.Rows[0] != exp {
t.Errorf("expected %f, got %+v", exp, item.Rows[0])
}
default:
t.Errorf("unknown type: %T", exp)
}
}
check := testCase.expect.([]interface{})
testExtractRowsQuery(t, m.Index, testCase.fieldName, check)
}
@ -568,58 +519,16 @@ id__ID,negneg__Int_-10_-5,negpos__Int_-10_10,pospos__Int_5_10,negzero__Int_-10_0
t.Run(fmt.Sprintf("batchsize=%d", bsize), func(t *testing.T) {
m := newMainOORFactory(t, file, true, false, false)
m.BatchSize = bsize
defer func() {
if err := m.PilosaClient().DeleteIndexByName(m.Index); err != nil {
t.Logf("deleting test index: %v", err)
}
}()
defer testDeleteIndex(t, m)
err := m.Run()
if err != nil {
t.Fatalf("running: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
for _, field := range []string{"negneg", "negzero", "negpos", "zeropos", "pospos"} {
pql := fmt.Sprintf("Extract(All(), Rows(%s))", field)
eResp, err := idktest.DoExtractQuery(pql, m.Index)
if err != nil {
t.Fatal("doing extract: ", err)
}
if eResp.Results[0].Columns == nil {
t.Fatal("no results: ", err)
}
for i, item := range eResp.Results[0].Columns {
check := checker[field].([]interface{})
switch exp := check[i].(type) {
case nil:
if item.Rows[0] != nil {
t.Errorf("expected nil, got %+v", item.Rows[0])
}
case string:
if item.Rows[0] != exp {
t.Errorf("expected %s, got %+v", exp, item.Rows[0])
}
case int:
vAsInt := int(item.Rows[0].(float64))
if vAsInt != exp {
t.Errorf("expected %d, got %+v", exp, item.Rows[0])
} else {
expAsFloat := float64(exp)
if item.Rows[0] != expAsFloat {
t.Errorf("expected %f, got %+v", expAsFloat, item.Rows[0])
}
}
case float64:
if item.Rows[0] != exp {
t.Errorf("expected %f, got %+v", exp, item.Rows[0])
}
default:
t.Errorf("unknown type: %T", exp)
}
}
check := checker[field].([]interface{})
testExtractRowsQuery(t, m.Index, field, check)
}
})
}
@ -651,44 +560,16 @@ id__ID,ts1__Timestamp_ns_2006-01-02 15:04:05.999,ts2__Timestamp_s_2006-01-02T15:
t.Run(fmt.Sprintf("batchsize=%d", bsize), func(t *testing.T) {
m := newMainOORFactory(t, file, false, false, true)
m.BatchSize = bsize
defer func() {
if err := m.PilosaClient().DeleteIndexByName(m.Index); err != nil {
t.Logf("deleting test index: %v", err)
}
}()
defer testDeleteIndex(t, m)
err := m.Run()
if err != nil {
t.Fatalf("running: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
for _, field := range []string{"ts1", "ts2", "ts3", "ts4"} {
pql := fmt.Sprintf("Extract(All(), Rows(%s))", field)
check := checker[field].([]interface{})
testExtractRowsQuery(t, m.Index, field, check)
eResp, err := idktest.DoExtractQuery(pql, m.Index)
if err != nil {
t.Fatal("doing extract: ", err)
}
if eResp.Results[0].Columns == nil {
t.Fatal("no results: ", err)
}
for i, item := range eResp.Results[0].Columns {
check := checker[field].([]interface{})
switch exp := check[i].(type) {
case nil:
if item.Rows[0] != nil {
t.Errorf("expected nil, got %+v", item.Rows[0])
}
case string:
if item.Rows[0] != exp {
t.Errorf("expected %s, got %+v", exp, item.Rows[0])
}
default:
t.Errorf("unknown type: %T", exp)
}
}
}
})
}
@ -743,27 +624,18 @@ func TestFailureConditions(t *testing.T) {
t.Run(test.name, func(t *testing.T) {
m := newMainOORFactory(t, test.csv, test.intOutOfRange, test.decimalOutOfRange, test.timestampOutOfRange)
m.BatchSize = 1
defer func() {
if err := m.PilosaClient().DeleteIndexByName(m.Index); err != nil {
t.Logf("deleting test index: %v", err)
}
}()
defer testDeleteIndex(t, m)
err := m.Run()
if test.fail {
if err == nil {
t.Fatal("expected error, got nil")
}
} else {
if err != nil {
t.Fatalf("running: %v", err)
}
} else if err != nil {
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
})
}
}
func newMainOORFactory(t *testing.T, file string, allowIntOOR bool, allowDecOOR bool, allowTSOOR bool) *Main {

574
idk/header.go
View file

@ -33,6 +33,8 @@ const (
var (
ErrNoFieldSpec = errors.New("no field spec in this header")
ErrInvalidFieldName = errors.New("field name must match [a-z][a-z0-9_-]{0,229}")
ErrParsingEpoch = "parsing epoch for "
ErrDecodingConfig = "decoding config for field "
)
// HeaderToField takes a header specification which looks like
@ -73,265 +75,31 @@ func HeaderToField(headerField string, log logger.Logger) (field Field, _ error)
switch fieldType {
case IDType:
idField := IDField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
if fieldspec[1] == "T" {
idField.Mutex = true
} else if fieldspec[1] != "F" {
return nil, errors.Errorf("can't interpret '%s' for IDField.Mutex for field '%s'", fieldspec[1], sourceName)
}
}
if len(fieldspec) > 2 {
idField.Quantum = fieldspec[2]
}
if len(fieldspec) > 3 {
idField.TTL = fieldspec[3]
}
if len(fieldspec) > 4 {
log.Printf("ignoring extra arguments to IDField %s: %v", headerField, fieldspec[4:])
}
field = idField
field, err = headerToIDField(headerField, sourceName, destName, fieldspec, log)
case BoolType:
field = BoolField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
log.Printf("ignoring extra arguments to BoolField %s: %v", headerField, fieldspec[1:])
}
field, err = headerToBoolField(headerField, sourceName, destName, fieldspec, log)
case StringType:
strField := StringField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
if fieldspec[1] == "T" {
strField.Mutex = true
} else if fieldspec[1] != "F" {
return nil, errors.Errorf("can't interpret '%s' for StringField.Mutex for field '%s'", fieldspec[1], sourceName)
}
}
if len(fieldspec) > 2 {
strField.Quantum = fieldspec[2]
}
if len(fieldspec) > 3 {
strField.TTL = fieldspec[3]
}
if len(fieldspec) > 4 {
log.Printf("ignoring extra arguments to StringField %s: %v", headerField, fieldspec[4:])
}
field = strField
field, err = headerToStringField(headerField, sourceName, destName, fieldspec, log)
case LookupTextType:
lTextField := LookupTextField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
log.Printf("ignoring extra arguments to LookupTextField %s: %v", headerField, fieldspec[1:])
}
field = lTextField
field, err = headerToLookupTextField(headerField, sourceName, destName, fieldspec, log)
case IntType:
intField := IntField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
min, err := strconv.ParseInt(fieldspec[1], 10, 64)
if err != nil {
return nil, errors.Wrapf(err, "parsing min for %s", sourceName)
}
intField.Min = &min
}
if len(fieldspec) > 2 {
max, err := strconv.ParseInt(fieldspec[2], 10, 64)
if err != nil {
return nil, errors.Wrapf(err, "parsing max for %s", sourceName)
}
intField.Max = &max
}
if len(fieldspec) > 3 {
intField.ForeignIndex = fieldspec[3]
}
if len(fieldspec) > 4 {
log.Printf("ignoring extra arguments to IntField %s: %v", headerField, fieldspec[4:])
}
field = intField
field, err = headerToIntField(headerField, sourceName, destName, fieldspec, log)
case ForeignKeyType:
fkField := IntField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
fkField.ForeignIndex = fieldspec[1]
} else {
return nil, errors.Errorf("need foreign index for foreign key field: %s", headerField)
}
if len(fieldspec) > 2 {
log.Printf("ignoring extra arguments to ForeignKey Field %s: %v", headerField, fieldspec[2:])
}
field = fkField
field, err = headerToForeignKeyField(headerField, sourceName, destName, fieldspec, log)
case DecimalType:
decField := DecimalField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
scale, err := strconv.ParseInt(fieldspec[1], 10, 64)
if err != nil {
return nil, errors.Wrapf(err, "parsing scale for %s", sourceName)
}
decField.Scale = scale
}
if len(fieldspec) > 2 {
log.Printf("ignoring extra arguments to DecimalField %s: %v", headerField, fieldspec[2:])
}
field = decField
field, err = headerToDecimalField(headerField, sourceName, destName, fieldspec, log)
case StringArrayType:
strArrField := StringArrayField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
strArrField.Quantum = fieldspec[1]
}
if len(fieldspec) > 2 {
strArrField.TTL = fieldspec[2]
}
if len(fieldspec) > 3 {
log.Printf("ignoring extra arguments to StringArrayField %s: %v", headerField, fieldspec[3:])
}
field = strArrField
field, err = headerToStringArrayField(headerField, sourceName, destName, fieldspec, log)
case IDArrayType:
idArrField := IDArrayField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
idArrField.Quantum = fieldspec[1]
}
if len(fieldspec) > 2 {
idArrField.TTL = fieldspec[2]
}
if len(fieldspec) > 3 {
log.Printf("ignoring extra arguments to IDArrayField %s: %v", headerField, fieldspec[3:])
}
field = idArrField
field, err = headerToIDArrayField(headerField, sourceName, destName, fieldspec, log)
case DateIntType:
dateField := DateIntField{
NameVal: sourceName,
DestNameVal: destName,
}
layout := time.RFC3339
if len(fieldspec) > 1 {
layout = fieldspec[1]
}
dateField.Layout = layout
if len(fieldspec) > 2 {
epoch, err := time.Parse(layout, fieldspec[2])
if err != nil {
return nil, errors.Wrapf(err, "parsing epoch for '%s'", headerField)
}
dateField.Epoch = epoch
}
if len(fieldspec) > 3 {
dateField.Unit = Unit(fieldspec[3]).unit()
if len(fieldspec) > 4 && (dateField.Unit.IsCustom()) {
if _, err := time.ParseDuration(fieldspec[4]); err != nil {
return nil, errors.Wrapf(err, "parsing custom unit %s", fieldspec[4])
}
dateField.CustomUnit = fieldspec[4]
} else {
if _, err := dateField.Unit.Duration(); err != nil {
return nil, err
}
}
}
if len(fieldspec) > 5 {
log.Printf("ignoring extra arguments to DateIntField %s: %v", headerField, fieldspec[5:])
}
field = dateField
field, err = headerToDateIntField(headerField, sourceName, destName, fieldspec, log)
case TimestampType:
tsField := TimestampField{
NameVal: sourceName,
DestNameVal: destName,
}
granularity := "s"
layout := time.RFC3339Nano
if len(fieldspec) > 1 {
granularity = fieldspec[1]
}
tsField.Granularity = granularity
if len(fieldspec) > 2 {
layout = fieldspec[2]
}
tsField.Layout = layout
if len(fieldspec) > 3 {
epoch, err := time.Parse(layout, fieldspec[3])
if err != nil {
return nil, errors.Wrapf(err, "parsing epoch for '%s'", headerField)
}
if epoch.IsZero() {
tsField.Epoch = time.Unix(0, 0)
} else {
tsField.Epoch = epoch
}
}
if len(fieldspec) > 4 {
unit := Unit(fieldspec[4]).unit()
if _, err := unit.Duration(); err != nil {
return nil, errors.Wrapf(err, "invalid unit for TimestampField %s", headerField)
}
tsField.Unit = unit
}
if len(fieldspec) > 5 {
log.Printf("ignoring extra arguments to TimestampField %s: %v", headerField, fieldspec[5:])
}
field = tsField
field, err = headerToTimestampField(headerField, sourceName, destName, fieldspec, log)
case RecordTimeType:
rtField := RecordTimeField{
NameVal: sourceName,
DestNameVal: destName,
}
// We used to use a default of "" here, and then
// (RecordTimeType).layout() would treat that as RFC3339.
// This is now more parallel to the handling for DateIntType.
layout := time.RFC3339
if len(fieldspec) > 1 {
layout = fieldspec[1]
}
rtField.Layout = layout
if len(fieldspec) > 2 {
epoch, err := time.Parse(layout, fieldspec[2])
if err != nil {
return nil, errors.Wrapf(err, "parsing epoch for '%s'", headerField)
}
rtField.Epoch = epoch
}
if len(fieldspec) > 3 {
unit := Unit(fieldspec[3]).unit()
_, err := unit.Duration()
if err != nil {
return nil, err
}
rtField.Unit = unit
}
if len(fieldspec) > 4 {
log.Printf("ignoring extra arguments to RecordTimeField %s: %v", headerField, fieldspec[4:])
}
field = rtField
field, err = headerToRecordTimeField(headerField, sourceName, destName, fieldspec, log)
case SignedIntBoolKeyType:
field = SignedIntBoolKeyField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
log.Printf("ignoring extra arguments to SignedIntBoolKeyField %s: %v", headerField, fieldspec[1:])
}
field, err = headerToSignedIntBoolKeyField(headerField, sourceName, destName, fieldspec, log)
case IgnoreType:
field = IgnoreField{}
if len(fieldspec) > 1 {
@ -341,6 +109,294 @@ func HeaderToField(headerField string, log logger.Logger) (field Field, _ error)
return nil, errors.Errorf("unknown field '%s' for '%s'", fieldspec[0], headerField)
}
return field, err
}
func headerToIDField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
idField := IDField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
if fieldspec[1] == "T" {
idField.Mutex = true
} else if fieldspec[1] != "F" {
return nil, errors.Errorf("can't interpret '%s' for IDField.Mutex for field '%s'", fieldspec[1], sourceName)
}
}
if len(fieldspec) > 2 {
idField.Quantum = fieldspec[2]
}
if len(fieldspec) > 3 {
idField.TTL = fieldspec[3]
}
if len(fieldspec) > 4 {
log.Printf("ignoring extra arguments to IDField %s: %v", headerField, fieldspec[4:])
}
return idField, nil
}
func headerToBoolField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
field := BoolField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
log.Printf("ignoring extra arguments to BoolField %s: %v", headerField, fieldspec[1:])
}
return field, nil
}
func headerToStringField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
strField := StringField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
if fieldspec[1] == "T" {
strField.Mutex = true
} else if fieldspec[1] != "F" {
return nil, errors.Errorf("can't interpret '%s' for StringField.Mutex for field '%s'", fieldspec[1], sourceName)
}
}
if len(fieldspec) > 2 {
strField.Quantum = fieldspec[2]
}
if len(fieldspec) > 3 {
strField.TTL = fieldspec[3]
}
if len(fieldspec) > 4 {
log.Printf("ignoring extra arguments to StringField %s: %v", headerField, fieldspec[4:])
}
return strField, nil
}
func headerToLookupTextField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
lTextField := LookupTextField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
log.Printf("ignoring extra arguments to LookupTextField %s: %v", headerField, fieldspec[1:])
}
return lTextField, nil
}
func headerToIntField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
intField := IntField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
min, err := strconv.ParseInt(fieldspec[1], 10, 64)
if err != nil {
return nil, errors.Wrapf(err, "parsing min for %s", sourceName)
}
intField.Min = &min
}
if len(fieldspec) > 2 {
max, err := strconv.ParseInt(fieldspec[2], 10, 64)
if err != nil {
return nil, errors.Wrapf(err, "parsing max for %s", sourceName)
}
intField.Max = &max
}
if len(fieldspec) > 3 {
intField.ForeignIndex = fieldspec[3]
}
if len(fieldspec) > 4 {
log.Printf("ignoring extra arguments to IntField %s: %v", headerField, fieldspec[4:])
}
return intField, nil
}
func headerToForeignKeyField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
fkField := IntField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
fkField.ForeignIndex = fieldspec[1]
} else {
return nil, errors.Errorf("need foreign index for foreign key field: %s", headerField)
}
if len(fieldspec) > 2 {
log.Printf("ignoring extra arguments to ForeignKey Field %s: %v", headerField, fieldspec[2:])
}
return fkField, nil
}
func headerToDecimalField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
decField := DecimalField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
scale, err := strconv.ParseInt(fieldspec[1], 10, 64)
if err != nil {
return nil, errors.Wrapf(err, "parsing scale for %s", sourceName)
}
decField.Scale = scale
}
if len(fieldspec) > 2 {
log.Printf("ignoring extra arguments to DecimalField %s: %v", headerField, fieldspec[2:])
}
return decField, nil
}
func headerToStringArrayField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
strArrField := StringArrayField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
strArrField.Quantum = fieldspec[1]
}
if len(fieldspec) > 2 {
strArrField.TTL = fieldspec[2]
}
if len(fieldspec) > 3 {
log.Printf("ignoring extra arguments to StringArrayField %s: %v", headerField, fieldspec[3:])
}
return strArrField, nil
}
func headerToIDArrayField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
idArrField := IDArrayField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
idArrField.Quantum = fieldspec[1]
}
if len(fieldspec) > 2 {
idArrField.TTL = fieldspec[2]
}
if len(fieldspec) > 3 {
log.Printf("ignoring extra arguments to IDArrayField %s: %v", headerField, fieldspec[3:])
}
return idArrField, nil
}
func headerToDateIntField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
dateField := DateIntField{
NameVal: sourceName,
DestNameVal: destName,
}
layout := time.RFC3339
if len(fieldspec) > 1 {
layout = fieldspec[1]
}
dateField.Layout = layout
if len(fieldspec) > 2 {
epoch, err := time.Parse(layout, fieldspec[2])
if err != nil {
return nil, errors.Wrapf(err, ErrParsingEpoch, headerField)
}
dateField.Epoch = epoch
}
if len(fieldspec) > 3 {
dateField.Unit = Unit(fieldspec[3]).unit()
if len(fieldspec) > 4 && (dateField.Unit.IsCustom()) {
if _, err := time.ParseDuration(fieldspec[4]); err != nil {
return nil, errors.Wrapf(err, "parsing custom unit %s", fieldspec[4])
}
dateField.CustomUnit = fieldspec[4]
} else {
if _, err := dateField.Unit.Duration(); err != nil {
return nil, err
}
}
}
if len(fieldspec) > 5 {
log.Printf("ignoring extra arguments to DateIntField %s: %v", headerField, fieldspec[5:])
}
return dateField, nil
}
func headerToTimestampField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
tsField := TimestampField{
NameVal: sourceName,
DestNameVal: destName,
}
granularity := "s"
layout := time.RFC3339Nano
if len(fieldspec) > 1 {
granularity = fieldspec[1]
}
tsField.Granularity = granularity
if len(fieldspec) > 2 {
layout = fieldspec[2]
}
tsField.Layout = layout
if len(fieldspec) > 3 {
epoch, err := time.Parse(layout, fieldspec[3])
if err != nil {
return nil, errors.Wrapf(err, ErrParsingEpoch, headerField)
}
if epoch.IsZero() {
tsField.Epoch = time.Unix(0, 0)
} else {
tsField.Epoch = epoch
}
}
if len(fieldspec) > 4 {
unit := Unit(fieldspec[4]).unit()
if _, err := unit.Duration(); err != nil {
return nil, errors.Wrapf(err, "invalid unit for TimestampField %s", headerField)
}
tsField.Unit = unit
}
if len(fieldspec) > 5 {
log.Printf("ignoring extra arguments to TimestampField %s: %v", headerField, fieldspec[5:])
}
return tsField, nil
}
func headerToRecordTimeField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
rtField := RecordTimeField{
NameVal: sourceName,
DestNameVal: destName,
}
// We used to use a default of "" here, and then
// (RecordTimeType).layout() would treat that as RFC3339.
// This is now more parallel to the handling for DateIntType.
layout := time.RFC3339
if len(fieldspec) > 1 {
layout = fieldspec[1]
}
rtField.Layout = layout
if len(fieldspec) > 2 {
epoch, err := time.Parse(layout, fieldspec[2])
if err != nil {
return nil, errors.Wrapf(err, ErrParsingEpoch, headerField)
}
rtField.Epoch = epoch
}
if len(fieldspec) > 3 {
unit := Unit(fieldspec[3]).unit()
_, err := unit.Duration()
if err != nil {
return nil, err
}
rtField.Unit = unit
}
if len(fieldspec) > 4 {
log.Printf("ignoring extra arguments to RecordTimeField %s: %v", headerField, fieldspec[4:])
}
return rtField, nil
}
func headerToSignedIntBoolKeyField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) {
field := SignedIntBoolKeyField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
log.Printf("ignoring extra arguments to SignedIntBoolKeyField %s: %v", headerField, fieldspec[1:])
}
return field, nil
}
@ -475,7 +531,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name
@ -486,7 +542,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name
@ -497,7 +553,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name
@ -508,7 +564,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name
@ -519,7 +575,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name
@ -530,7 +586,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name
@ -541,7 +597,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name
@ -552,7 +608,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name
@ -563,7 +619,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name
@ -574,7 +630,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name
@ -585,7 +641,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name
@ -595,7 +651,7 @@ func ParseHeader(raw []byte) ([]Field, PathTable, error) {
if s.Config != nil {
err := json.Unmarshal(s.Config, &field)
if err != nil {
return nil, nil, errors.Wrapf(err, "decoding config for field %q", s.Name)
return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name)
}
}
field.NameVal = s.Name

7
idk/idktest/idktest.go
View file

@ -11,6 +11,13 @@ import (
"github.com/pkg/errors"
)
const (
ErrDeletingIndex = "deleting index"
ErrRunningIngest = "running ingester"
ErrGettingSchema = "getting schema"
ErrGettingQuery = "querying"
)
type ExtractResponse struct {
Results []Result
}

24
idk/ingest.go
View file

@ -42,8 +42,11 @@ import (
type contextKey int
const contextKeyToken contextKey = iota
const (
contextKeyToken contextKey = iota
Exists = "-exists"
ErrCommittingIDs = "committing IDs for batch"
)
// TODO Jaeger
@ -378,7 +381,7 @@ initialFetch:
if nexter != nil {
err := nexter.Commit(ctx)
if err != nil {
return errors.Wrap(err, "committing IDs for batch")
return errors.Wrap(err, ErrCommittingIDs)
}
}
m.log.Printf("imported batch after timeout")
@ -413,7 +416,7 @@ initialFetch:
if nexter != nil {
ierr := nexter.Commit(ctx)
if ierr != nil {
return errors.Wrap(ierr, "committing IDs for batch")
return errors.Wrap(ierr, ErrCommittingIDs)
}
}
m.log.Printf("1 records processed %v-> (%v)", sourceInstance, recordCounter)
@ -534,7 +537,7 @@ initialFetch:
if nexter != nil {
err = nexter.Commit(ctx)
if err != nil {
return errors.Wrap(err, "committing IDs for batch")
return errors.Wrap(err, ErrCommittingIDs)
}
}
batchStart = true
@ -1111,7 +1114,7 @@ func (m *Main) runDeleter(c int, limitCounter *msgCounter) error {
return errors.Errorf("unsupported directive '%s' field name must be equal to packed bools field: '%s'", directive, m.PackBools)
}
boolsField := index.Field(m.PackBools)
boolsExists := index.Field(m.PackBools + "-exists")
boolsExists := index.Field(m.PackBools + Exists)
_, err := client.Query(index.BatchQuery(
boolsField.Clear(value, recordID),
boolsExists.Clear(value, recordID),
@ -1418,7 +1421,7 @@ func (m *Main) batchFromSchema(schema []Field) ([]Recordizer, pilosaclient.Recor
return nil, nil, nil, nil, errors.Wrap(err, "checking packed bool field compatibility")
}
}
packBoolsExistsFld := m.PackBools + "-exists"
packBoolsExistsFld := m.PackBools + Exists
if m.index.HasField(packBoolsExistsFld) {
pBoolFieldExists := m.index.Field(packBoolsExistsFld)
if err := m.checkFieldCompatibility(pBoolFieldExists, nil, packBoolsExistsFld); err != nil {
@ -1426,7 +1429,7 @@ func (m *Main) batchFromSchema(schema []Field) ([]Recordizer, pilosaclient.Recor
}
}
boolField = m.index.Field(m.PackBools, pilosaclient.OptFieldTypeSet(pilosaclient.CacheTypeRanked, pilosacore.DefaultCacheSize), pilosaclient.OptFieldKeys(true))
boolFieldExists = m.index.Field(m.PackBools+"-exists", pilosaclient.OptFieldTypeSet(pilosaclient.CacheTypeRanked, pilosacore.DefaultCacheSize), pilosaclient.OptFieldKeys(true))
boolFieldExists = m.index.Field(m.PackBools+Exists, pilosaclient.OptFieldTypeSet(pilosaclient.CacheTypeRanked, pilosacore.DefaultCacheSize), pilosaclient.OptFieldKeys(true))
}
fields := make([]*pilosaclient.Field, 0, len(schema))
@ -1661,7 +1664,7 @@ func (m *Main) batchFromSchema(schema []Field) ([]Recordizer, pilosaclient.Recor
name := fld.DestName()
fields = append(fields,
m.index.Field(name, pilosaclient.OptFieldTypeSet(pilosaclient.CacheTypeRanked, pilosacore.DefaultCacheSize)),
m.index.Field(name+"-exists", pilosaclient.OptFieldTypeSet(pilosaclient.CacheTypeRanked, pilosacore.DefaultCacheSize)),
m.index.Field(name+Exists, pilosaclient.OptFieldTypeSet(pilosaclient.CacheTypeRanked, pilosacore.DefaultCacheSize)),
)
valIdx := len(fields) - 2
recordizers = append(recordizers, func(rawRec []interface{}, rec *pilosaclient.Row) (err error) {
@ -1973,10 +1976,7 @@ func getPrimaryKeyRecordizer(schema []Field, pkFields []string) (recordizer Reco
// primary key field and it is a byte slice already.
if len(fieldIndices) == 1 {
switch recID := rawRec[fieldIndices[0]].(type) {
case []byte:
rec.ID = recID
return nil
case string:
case []byte, string:
rec.ID = recID
return nil
}

93
idk/ingest_test.go
View file

@ -18,6 +18,7 @@ import (
"github.com/golang-jwt/jwt"
"github.com/molecula/featurebase/v3/authn"
pilosaclient "github.com/molecula/featurebase/v3/client"
"github.com/molecula/featurebase/v3/idk/idktest"
"github.com/molecula/featurebase/v3/logger"
"github.com/pkg/errors"
)
@ -60,7 +61,7 @@ func TestErrFlush(t *testing.T) {
err := ingester.Run()
if err != nil {
t.Fatalf("running ingester: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
client := ingester.PilosaClient()
@ -102,7 +103,7 @@ func TestErrBatchNowStale(t *testing.T) {
defer func() {
if err := ingester.PilosaClient().DeleteIndexByName(ingester.Index); err != nil {
t.Logf("deleting test index: %v", err)
t.Logf("%s for index %s: %v", idktest.ErrDeletingIndex, ingester.Index, err)
}
}()
@ -110,7 +111,7 @@ func TestErrBatchNowStale(t *testing.T) {
go func() {
err := ingester.Run()
if err != nil {
t.Logf("running ingester: %v", err)
t.Logf("%s: %v", idktest.ErrRunningIngest, err)
}
close(signal)
}()
@ -154,13 +155,13 @@ func TestIngestSignedIntBoolField(t *testing.T) {
configureTestFlags(ingester)
ingester.NewSource = func() (Source, error) { return ts, nil }
rand.Seed(time.Now().UTC().UnixNano())
ingester.Index = fmt.Sprintf("satest%d", rand.Intn(100000))
ingester.Index = fmt.Sprintf("ingestint%d", rand.Intn(100000))
ingester.BatchSize = 2
ingester.PrimaryKeyFields = []string{"rcid"}
err := ingester.Run()
if err != nil {
t.Fatalf("running ingester: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
client := ingester.PilosaClient()
@ -173,34 +174,24 @@ func TestIngestSignedIntBoolField(t *testing.T) {
svals := ingester.index.Field("svals")
svalsex := ingester.index.Field("svals-exists")
if resp, err := client.Query(svals.Row(22)); err != nil {
t.Fatalf("row 22: %v", err)
} else if !stringSliceSame(resp.ResultList[0].Row().Keys, []string{"b"}) {
t.Fatalf("wanted [b], got: %+v", resp.ResultList[0].Row().Keys)
tests := []struct {
field *pilosaclient.Field
row int
want []string
}{
{field: svals, row: 22, want: []string{"b"}},
{field: svalsex, row: 22, want: []string{"a", "b"}},
{field: svalsex, row: 44, want: []string{"a", "b"}},
{field: svalsex, row: 5, want: []string{"a", "c"}},
{field: svals, row: 5, want: []string{"a", "c"}},
}
if resp, err := client.Query(svalsex.Row(22)); err != nil {
t.Fatalf("row 22: %v", err)
} else if !stringSliceSame(resp.ResultList[0].Row().Keys, []string{"a", "b"}) {
t.Fatalf("wanted [a b], got: %+v", resp.ResultList[0].Row().Keys)
}
if resp, err := client.Query(svalsex.Row(44)); err != nil {
t.Fatalf("row 22: %v", err)
} else if !stringSliceSame(resp.ResultList[0].Row().Keys, []string{"a", "b"}) {
t.Fatalf("wanted [b], got: %+v", resp.ResultList[0].Row().Keys)
}
if resp, err := client.Query(svalsex.Row(5)); err != nil {
t.Fatalf("row 22: %v", err)
} else if !stringSliceSame(resp.ResultList[0].Row().Keys, []string{"a", "c"}) {
t.Fatalf("wanted [b], got: %+v", resp.ResultList[0].Row().Keys)
}
if resp, err := client.Query(svals.Row(5)); err != nil {
t.Fatalf("row 22: %v", err)
} else if !stringSliceSame(resp.ResultList[0].Row().Keys, []string{"a", "c"}) {
t.Fatalf("wanted [b], got: %+v", resp.ResultList[0].Row().Keys)
for _, test := range tests {
if resp, err := client.Query(test.field.Row(test.row)); err != nil {
t.Fatalf("row %d: %v", test.row, err)
} else if !stringSliceSame(resp.ResultList[0].Row().Keys, test.want) {
t.Fatalf("wanted %+v, got: %+v", test.want, resp.ResultList[0].Row().Keys)
}
}
}
@ -223,7 +214,7 @@ func TestSingleBoolClear(t *testing.T) {
ingester.IDField = "id"
if err := ingester.Run(); err != nil {
t.Fatalf("running ingester: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
client := ingester.PilosaClient()
@ -281,7 +272,7 @@ func TestForeignKeyRegression(t *testing.T) {
defer func() {
if err := ingester.PilosaClient().DeleteIndexByName(ingester.Index); err != nil {
t.Logf("deleting test index: %v", err)
t.Logf("%s for index %s: %v", idktest.ErrDeletingIndex, ingester.Index, err)
}
}()
@ -304,7 +295,7 @@ func TestForeignKeyRegression(t *testing.T) {
defer func() {
if err := client.DeleteIndexByName("testusers834"); err != nil {
t.Logf("deleting testusers index: %v", err)
t.Logf("%s for index %s: %v", idktest.ErrDeletingIndex, "testusers834", err)
}
}()
@ -312,7 +303,7 @@ func TestForeignKeyRegression(t *testing.T) {
// getting cleared for int fields.
err = ingester.run()
if err != nil {
t.Fatalf("running ingester: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
}
@ -384,16 +375,16 @@ func TestIngestStringArrays(t *testing.T) {
ingester := tt.ingester()
ingester.NewSource = func() (Source, error) { return &tt.source, nil }
rand.Seed(time.Now().UTC().UnixNano())
ingester.Index = fmt.Sprintf("satest%d", rand.Intn(100000))
ingester.Index = fmt.Sprintf("ingeststring%d", rand.Intn(100000))
ingester.BatchSize = 5
defer func() {
if err := ingester.PilosaClient().DeleteIndexByName(ingester.Index); err != nil {
t.Logf("deleting test index: %v", err)
t.Logf("%s for index %s: %v", idktest.ErrDeletingIndex, ingester.index, err)
}
}()
err := ingester.Run()
if err != nil {
t.Fatalf("running ingester: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
client := ingester.PilosaClient()
@ -433,7 +424,7 @@ func TestIngesterServesPrometheusEndpoint(t *testing.T) {
go func() {
err := ingester.Run()
if err != nil {
t.Logf("running ingester: %v", err)
t.Logf("%s: %v", idktest.ErrRunningIngest, err)
}
close(signal)
}()
@ -461,10 +452,10 @@ func TestIngesterServesPrometheusEndpoint(t *testing.T) {
func TestDelete(t *testing.T) {
rand.Seed(time.Now().UTC().UnixNano())
indexName := fmt.Sprintf("satest%d", rand.Intn(100000))
indexName := fmt.Sprintf("delete%d", rand.Intn(100000))
primaryKeyFields := []string{"aba", "db", "id"}
ts_write := newTestSource(
tsWrite := newTestSource(
[]Field{
StringField{NameVal: "aba"},
StringField{NameVal: "db"},
@ -483,7 +474,7 @@ func TestDelete(t *testing.T) {
)
// first set up deleter - we want to make sure that even if it starts first before the index is created that things still work
ts_delete := newTestSource(
tsDelete := newTestSource(
[]Field{
//IDField{NameVal: "id"},
StringField{NameVal: "aba"},
@ -518,7 +509,7 @@ func TestDelete(t *testing.T) {
deleter := NewMain()
configureTestFlags(deleter)
deleter.Delete = true
deleter.NewSource = func() (Source, error) { return ts_delete, nil }
deleter.NewSource = func() (Source, error) { return tsDelete, nil }
deleter.Index = indexName
deleter.BatchSize = 5
deleter.PrimaryKeyFields = []string{"aba", "db", "id"}
@ -531,20 +522,20 @@ func TestDelete(t *testing.T) {
ingester := NewMain()
configureTestFlags(ingester)
ingester.NewSource = func() (Source, error) { return ts_write, nil }
ingester.NewSource = func() (Source, error) { return tsWrite, nil }
ingester.PrimaryKeyFields = primaryKeyFields
ingester.Index = indexName
ingester.BatchSize = 1
defer func() {
if err := ingester.PilosaClient().DeleteIndexByName(ingester.Index); err != nil {
t.Logf("deleting test index: %v", err)
t.Logf("%s for index %s: %v", idktest.ErrDeletingIndex, ingester.Index, err)
}
}()
err = ingester.Run()
if err != nil {
t.Fatalf("running ingester: %v", err)
t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
}
client := ingester.PilosaClient()
@ -725,7 +716,7 @@ func TestBatchFromSchema(t *testing.T) {
defer func() {
err := m.client.DeleteIndex(m.index)
if err != nil {
t.Logf("deleting test index: %v", err)
t.Logf("%s for index %s: %v", idktest.ErrDeletingIndex, m.Index, err)
}
}()
}
@ -1076,7 +1067,7 @@ func TestCheckFieldCompatibility(t *testing.T) {
}
defer func() {
if err := ingester.PilosaClient().DeleteIndexByName(idxName); err != nil {
t.Logf("deleting test index: %v", err)
t.Logf("%s for index %s: %v", idktest.ErrDeletingIndex, idxName, err)
}
}()
if err := client.CreateField(idx.Field("pset", pilosaclient.OptFieldTypeSet(pilosaclient.CacheTypeNone, 0))); err != nil {
@ -1339,7 +1330,7 @@ func getAuthToken(t *testing.T) string {
return token
}
func Test_Setup(t *testing.T) {
func TestSetup(t *testing.T) {
m := NewMain()
configureTestFlags(m)
m.AutoGenerate = true
@ -1362,7 +1353,7 @@ func Test_Setup(t *testing.T) {
}
})
}
func Test_NilIngest(t *testing.T) {
func TestNilIngest(t *testing.T) {
type testcase struct {
name string
schema []Field
@ -1406,7 +1397,7 @@ func Test_NilIngest(t *testing.T) {
defer func() {
err := m.client.DeleteIndex(m.index)
if err != nil {
t.Logf("deleting test index: %v", err)
t.Logf("%s for index %s: %v", idktest.ErrDeletingIndex, m.Index, err)
}
}()
}

17
idk/interfaces_test.go
View file

@ -240,7 +240,8 @@ func TestTTLOf(t *testing.T) {
}
func TestTimestampField_validateDuration(t *testing.T) {
func TestTimestampFieldValidateDuration(t *testing.T) {
errTimeOutsideEpoch := "value + epoch is too far from Unix epoch"
type args struct {
dur int64
offset int64
@ -269,7 +270,7 @@ func TestTimestampField_validateDuration(t *testing.T) {
unit: Unit("s"),
},
wantErr: true,
errStr: "value + epoch is too far from Unix epoch",
errStr: errTimeOutsideEpoch,
},
{
name: "max-oor",
@ -279,7 +280,7 @@ func TestTimestampField_validateDuration(t *testing.T) {
unit: Unit("s"),
},
wantErr: true,
errStr: "value + epoch is too far from Unix epoch",
errStr: errTimeOutsideEpoch,
},
{
name: "min-min",
@ -298,7 +299,7 @@ func TestTimestampField_validateDuration(t *testing.T) {
unit: Unit("s"),
},
wantErr: true,
errStr: "value + epoch is too far from Unix epoch",
errStr: errTimeOutsideEpoch,
},
{
name: "min-oor",
@ -308,7 +309,7 @@ func TestTimestampField_validateDuration(t *testing.T) {
unit: Unit("s"),
},
wantErr: true,
errStr: "value + epoch is too far from Unix epoch",
errStr: errTimeOutsideEpoch,
},
{
name: "max-max",
@ -327,7 +328,7 @@ func TestTimestampField_validateDuration(t *testing.T) {
unit: Unit("ns"),
},
wantErr: true,
errStr: "value + epoch is too far from Unix epoch",
errStr: errTimeOutsideEpoch,
},
{
name: "max-oor",
@ -337,7 +338,7 @@ func TestTimestampField_validateDuration(t *testing.T) {
unit: Unit("ns"),
},
wantErr: true,
errStr: "value + epoch is too far from Unix epoch",
errStr: errTimeOutsideEpoch,
},
}
for _, tt := range tests {
@ -351,7 +352,7 @@ func TestTimestampField_validateDuration(t *testing.T) {
}
}
func Test_validateTimestamp(t *testing.T) {
func TestValidateTimestamp(t *testing.T) {
type args struct {
unit Unit
ts time.Time