diff --git a/idk/csv/csv_test.go b/idk/csv/csv_test.go index 17f8198b5..a4c822204 100644 --- a/idk/csv/csv_test.go +++ b/idk/csv/csv_test.go @@ -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 { diff --git a/idk/header.go b/idk/header.go index 17abf65e4..cb0e37188 100644 --- a/idk/header.go +++ b/idk/header.go @@ -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 diff --git a/idk/idktest/idktest.go b/idk/idktest/idktest.go index 826db409c..8ba55f0e6 100644 --- a/idk/idktest/idktest.go +++ b/idk/idktest/idktest.go @@ -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 } diff --git a/idk/ingest.go b/idk/ingest.go index cbd5d7c50..85dc85793 100644 --- a/idk/ingest.go +++ b/idk/ingest.go @@ -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 } diff --git a/idk/ingest_test.go b/idk/ingest_test.go index 4d2f5ecd8..55c8612b5 100644 --- a/idk/ingest_test.go +++ b/idk/ingest_test.go @@ -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) } }() } diff --git a/idk/interfaces_test.go b/idk/interfaces_test.go index c5b4a2002..069cc54ae 100644 --- a/idk/interfaces_test.go +++ b/idk/interfaces_test.go @@ -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