From bb1d52a385b987a8eec8fc3ef1f9d55073faf6be Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 2 Aug 2021 16:33:08 -0500 Subject: [PATCH] ingest and ingest/codec testing work This is a design to let us write test cases for ingest with schema setup and data in the json formats we want to use, and results as alternating queries and expected results, so we can just create new test files and run the tests against them. We also have to report back what we created when creating things. In the process of developing this, I noticed that the documentation describes ingest schema as allowing more than one schema operation, but we didn't support this, and also it wouldn't do much good because there was no way to do partial things like "just add a field". Fixed. Also we implement comparison for ops, so the test output is actually a test rather than just some data to visually eyeball. In the process, realize that the handling of timestamps was wrong; we said that we take them as raw numbers relative to the epoch, not as raw Unix timestamps. Also a couple of related cleanups caught by doing the testing. --- api.go | 14 +- api_test.go | 28 +- fragment.go | 8 +- http/client.go | 45 ++- http/handler.go | 276 ++++++++++++------ http/handler_test.go | 1 + ingest/codec.go | 14 +- ingest/codec_test.go | 195 ++++++++++--- ingest/op.go | 112 +++++++- ingest/op_test.go | 2 +- ingest_test.go | 468 +++++++++++++++++++++++++++++++ ingest_testdata/bool.tc | 17 ++ ingest_testdata/expect_errors.tc | 101 +++++++ ingest_testdata/keyed.tc | 32 +++ ingest_testdata/sample.tc | 89 ++++++ tracker.go | 2 +- 16 files changed, 1254 insertions(+), 150 deletions(-) create mode 100644 ingest_test.go create mode 100644 ingest_testdata/bool.tc create mode 100644 ingest_testdata/expect_errors.tc create mode 100644 ingest_testdata/keyed.tc create mode 100644 ingest_testdata/sample.tc diff --git a/api.go b/api.go index 56a226698..aa7cdc79a 100644 --- a/api.go +++ b/api.go @@ -1863,7 +1863,7 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string } case "time": if err = codec.AddTimeQuantumField(field.name, lookup); err != nil { - return fmt.Errorf("adding time field to codec: %w", err) + return fmt.Errorf("adding time quantum field to codec: %w", err) } case "mutex": if err = codec.AddMutexField(field.name, lookup); err != nil { @@ -1871,20 +1871,20 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string } case "bool": if err = codec.AddBoolField(field.name); err != nil { - return fmt.Errorf("adding mutex field to codec: %w", err) + return fmt.Errorf("adding bool field to codec: %w", err) } case "int": if err = codec.AddIntField(field.name, lookup); err != nil { - return fmt.Errorf("adding mutex field to codec: %w", err) + return fmt.Errorf("adding int field to codec: %w", err) } case "decimal": if err = codec.AddDecimalField(field.name, field.options.Scale); err != nil { - return fmt.Errorf("adding mutex field to codec: %w", err) + return fmt.Errorf("adding decimal field to codec: %w", err) } case "timestamp": nanos := TimeUnitNanos(field.options.TimeUnit) if err = codec.AddTimestampField(field.name, time.Duration(nanos), field.options.Base); err != nil { - return fmt.Errorf("adding mutex field to codec: %w", err) + return fmt.Errorf("adding timestamp field to codec: %w", err) } default: return fmt.Errorf("unhandled field type %q", field.Type()) @@ -1894,7 +1894,7 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string if err != nil { return errors.Wrap(err, "parsing input data") } - sharded, err := req.Shard() + sharded, err := req.ByShard() if err != nil { return errors.Wrap(err, "sharding input data") } @@ -1983,7 +1983,7 @@ func (api *API) applyOperations(ctx context.Context, qcx *Qcx, index *Index, sha return errors.Wrap(err, "importing existence columns") } switch field.Type() { - case "set", "time", "mutex": + case "set", "time", "mutex", "bool": err = field.Import(qcx, fieldOp.Values, fieldOp.RecordIDs, fieldOp.Signed, shard, funcOpts...) case "int", "timestamp", "decimal": err = field.importValue(qcx, fieldOp.RecordIDs, fieldOp.Signed, shard, opts) diff --git a/api_test.go b/api_test.go index 17c1761d8..7e34f5be4 100644 --- a/api_test.go +++ b/api_test.go @@ -17,10 +17,10 @@ package pilosa_test import ( "bytes" "context" - "crypto/rand" "errors" "fmt" "math" + "math/rand" "reflect" "strings" "testing" @@ -511,17 +511,27 @@ func TestAPI_Ingest(t *testing.T) { } } -func BenchmarkIngest(b *testing.B) { - b.StopTimer() +// ingestBenchmarkHelper makes it easier to exclude this from benchmark computations +// and profiles. +func ingestBenchmarkHelper() []byte { buf := &bytes.Buffer{} buf.WriteString(`[{"action": "write", "records": {`) comma := "" + now := time.Now().Add(-3840000 * time.Second) for i := 0; i < 1000000; i++ { - fmt.Fprintf(buf, `%s"%d": { "set": [%d, %d] }`, comma, i, i%2, (i%4)+2) + then := now.Add(time.Duration(rand.Int63n(1234567)) * time.Second) + fmt.Fprintf(buf, `%s"%d": { "set": [%d, %d], "int": %d, "tq": { "time": "%s", "values": %d } }`, comma, i, i%2, (i%4)+2, rand.Int63n(163840), + then.Format(time.RFC3339), rand.Int63n(25)) comma = ", " } buf.WriteString(`}}]`) data := buf.Bytes() + return data +} + +func BenchmarkIngest(b *testing.B) { + b.StopTimer() + data := ingestBenchmarkHelper() ctx, cancel := context.WithCancel(context.Background()) defer cancel() c := test.MustRunCluster(b, 1, @@ -541,6 +551,8 @@ func BenchmarkIngest(b *testing.B) { index := "ingest" setField := "set" + intField := "int" + tqField := "tq" _, err := coord.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: false}) if err != nil { b.Fatalf("creating index: %v", err) @@ -549,6 +561,14 @@ func BenchmarkIngest(b *testing.B) { if err != nil { b.Fatalf("creating field: %v", err) } + _, err = coord.API.CreateField(ctx, index, intField, pilosa.OptFieldTypeInt(0, 163840)) + if err != nil { + b.Fatalf("creating field: %v", err) + } + _, err = coord.API.CreateField(ctx, index, tqField, pilosa.OptFieldTypeTime("YMDH")) + if err != nil { + b.Fatalf("creating field: %v", err) + } b.ReportAllocs() b.StartTimer() for i := 0; i < b.N; i++ { diff --git a/fragment.go b/fragment.go index 23f312627..844cc41df 100644 --- a/fragment.go +++ b/fragment.go @@ -2538,10 +2538,10 @@ func (f *fragment) bulkImportMutex(tx Tx, rowIDs, columnIDs []uint64, options *I // pruned, (2) the input is sorted by row IDs and then column IDs, // meaning that we will generate positions in strictly sequential order. if !options.fullySorted { - p := parallelSlices{cols: columnIDs, rows: rowIDs} - p.fullPrune() - columnIDs = p.cols - rowIDs = p.rows + p := parallelSlices{cols: columnIDs, rows: rowIDs} + p.fullPrune() + columnIDs = p.cols + rowIDs = p.rows } // create a mask of columns we care about diff --git a/http/client.go b/http/client.go index 509ade5ee..000fe5f2a 100644 --- a/http/client.go +++ b/http/client.go @@ -201,15 +201,21 @@ func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error return rsp.Indexes, nil } -// IngestSchema uses the new schema ingest endpoint. -func (c *InternalClient) IngestSchema(ctx context.Context, uri *pnet.URI, buf []byte) error { +// IngestSchema uses the new schema ingest endpoint. It returns a +// map from index names to fields created within them; note that if the +// entire index was created, the list of fields is empty. The intended +// usage is cleaning up after creating the indexes, so if you create the +// index, you don't need to delete the fields, but if you created fields +// within an existing index, you should delete those fields but not the +// whole index. +func (c *InternalClient) IngestSchema(ctx context.Context, uri *pnet.URI, buf []byte) (created map[string][]string, err error) { if uri == nil { uri = c.defaultURI } u := uri.Path("/internal/schema") req, err := http.NewRequest("POST", u, bytes.NewReader(buf)) if err != nil { - return errors.Wrap(err, "creating request") + return nil, errors.Wrap(err, "creating request") } req.Header.Set("Content-Length", strconv.Itoa(len(buf))) @@ -217,15 +223,38 @@ func (c *InternalClient) IngestSchema(ctx context.Context, uri *pnet.URI, buf [] req.Header.Set("Accept", "application/json") req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) - resp, err := c.executeRequest(req.WithContext(ctx)) + resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true)) if err != nil { - return errors.Wrap(err, "executing request") + return nil, errors.Wrap(err, "executing request") } defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - return errors.Errorf("unexpected status code: %s", resp.Status) + buf, err = ioutil.ReadAll(resp.Body) + if resp.StatusCode != 200 { + if err != nil { + return nil, errors.Wrapf(err, "bad status '%s' and err reading body", resp.Status) + } + var msg string + // try to decode a JSON response + var sr successResponse + qr := &pilosa.QueryResponse{} + if err = json.Unmarshal(buf, &sr); err == nil { + msg = sr.Error.Error() + } else if err := c.serializer.Unmarshal(buf, qr); err == nil { + msg = qr.Err.Error() + } else { + msg = string(buf) + } + return nil, errors.Errorf("against %s %s: '%s'", req.URL.String(), resp.Status, msg) } - return nil + // this is the err from ioutil.ReadAll, but in the case where resp.StatusCode + // was 2xx, so we don't have a bad status. + if err != nil { + return nil, errors.Wrapf(err, "error reading response body") + } + if err = json.Unmarshal(buf, &created); err != nil { + return nil, errors.Wrapf(err, "error interpreting response body") + } + return created, nil } // IngestOperations uses the new ingest endpoint for ingest data diff --git a/http/handler.go b/http/handler.go index 50476f862..663ccb0b9 100644 --- a/http/handler.go +++ b/http/handler.go @@ -581,7 +581,6 @@ func (r *successResponse) check(err error) (statusCode int) { } r.Success = false - fmt.Printf("err: %#v\n", err) r.Error = &Error{Message: err.Error()} return statusCode @@ -1357,7 +1356,8 @@ func (h *Handler) handleIngestData(w http.ResponseWriter, r *http.Request) { type ingestSpec struct { IndexName string `json:"index-name"` - IfNotExists bool `json:"if-not-exists"` + IndexAction string `json:"index-action"` + FieldAction string `json:"field-action"` PrimaryKeyType string `json:"primary-key-type"` Fields []fieldSpec `json:"fields"` } @@ -1413,6 +1413,155 @@ func fieldSpecToFieldOption(fSpec fieldSpec) fieldOptions { return opt } +// applyOneIngestSchema applies a single ingestSpec, which specifies operations on +// a single index and possibly fields. If it is successful, it returns the name +// of the index and an empty slice (if it created the index), or the name of the +// index and a slice of the fields within that index that it created. If it +// is unsuccessful, it tries to delete whatever it created. +// +// The intended idiom is that if the returned list of fields isn't empty, the index +// already existed and only those fields need to be cleaned up in the event of +// a later error, but if the list of fields is empty, the entire index was new, +// and should be cleaned up, in which case there's no need to track or delete +// the specific fields separately. +func (h *Handler) applyOneIngestSchema(ctx context.Context, schema *ingestSpec) (index *pilosa.Index, returnedFields []string, err error) { + // create index + indexName := schema.IndexName + var createdFields []string + var useKeys bool + switch schema.PrimaryKeyType { + case "string": + useKeys = true + case "uint": + useKeys = false + default: + return nil, nil, fmt.Errorf("invalid primary key type %q", schema.PrimaryKeyType) + } + opts := pilosa.IndexOptions{ + Keys: useKeys, + TrackExistence: true, + } + createdIndex := false + + // We check this up here because, if there's at least one field but we don't know what to do with + // it, we will necessarily fail, which means we'd delete the index anyway, so there's no point in + // trying to create it. We don't care about this if there's no fields specified. + if len(schema.Fields) > 0 { + switch schema.FieldAction { + case "create", "ensure", "require": + // do nothing + case "": + schema.FieldAction = schema.IndexAction + default: + return nil, nil, fmt.Errorf("invalid field-action %q, expecting create/ensure/require", schema.FieldAction) + } + } + + switch schema.IndexAction { + case "ensure", "require": + index, err = h.api.Index(ctx, indexName) + if err != nil { + if _, ok := err.(pilosa.NotFoundError); !ok { + return nil, nil, fmt.Errorf("checking for existing index %q: %w", indexName, err) + } else { + err = nil + } + } + if index != nil { + existingOpts := index.Options() + if existingOpts != opts { + return nil, nil, fmt.Errorf("index %q options mismatch: schema %#v, existing %#v", indexName, opts, existingOpts) + } + break + } + if schema.IndexAction == "require" { + return nil, nil, fmt.Errorf("index %q does not exist", indexName) + } + fallthrough + case "create": + index, err = h.api.CreateIndex(ctx, indexName, opts) + if err != nil { + return nil, nil, err + } + createdIndex = true + default: + return nil, nil, fmt.Errorf("invalid index-action %q, need create/ensure/require", schema.IndexAction) + } + + // Now we might have an index, so we need our cleanup code. + defer func() { + if err == nil { + return + } + if createdIndex { + err := h.api.DeleteIndex(ctx, indexName) + if err != nil { + h.logger.Printf("trying to undo failed index %q creation: %v", indexName, err) + } + return + } + for _, field := range createdFields { + err := h.api.DeleteField(ctx, indexName, field) + if err != nil { + h.logger.Printf("trying to undo failed field %q creation in index %q: %v", field, indexName, err) + } + } + }() + + // create all the fields specified in the index + for _, fSpec := range schema.Fields { + fieldName := fSpec.FieldName + opt := fieldSpecToFieldOption(fSpec) + err = opt.validate() + if err != nil { + return nil, nil, err + } + switch schema.FieldAction { + case "ensure", "require": + field, schemaErr := h.api.Field(ctx, indexName, fieldName) + if schemaErr != nil { + // NotFoundError is fine + if _, ok := schemaErr.(pilosa.NotFoundError); !ok { + return nil, nil, fmt.Errorf("checking for existing field %q in %q: %w", fieldName, indexName, err) + } + } + if field != nil { + existing := field.Options() + if opt.Type != existing.Type { + return nil, nil, fmt.Errorf("existing field %q is %q, not %q", fieldName, existing.Type, opt.Type) + } + if ((opt.Keys != nil) && *opt.Keys) != existing.Keys { + if existing.Keys { + return nil, nil, fmt.Errorf("existing field %q in %q uses keys", fieldName, indexName) + } else { + return nil, nil, fmt.Errorf("existing field %q in %q doesn't use keys", fieldName, indexName) + } + } + // TODO: verify compatibility of other field opts, this is sorta hard + break + } + if schema.FieldAction == "require" { + return nil, nil, fmt.Errorf("field %q does not exist in %q", fieldName, indexName) + } + fallthrough + case "create": + fos := fieldOptionsToFunctionalOpts(opt) + _, err = h.api.CreateField(ctx, indexName, fieldName, fos...) + if err != nil { + return nil, nil, fmt.Errorf("creating field %q in %q: %v", fieldName, indexName, err) + } + createdFields = append(createdFields, fieldName) + } + } + // we don't report the fields back, so we can distinguish "created index" + // from "created fields within index" + if createdIndex { + createdFields = nil + } + + return index, createdFields, nil +} + func (h *Handler) handleIngestSchema(w http.ResponseWriter, r *http.Request) { if !validHeaderAcceptJSON(r.Header) { http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) @@ -1424,93 +1573,60 @@ func (h *Handler) handleIngestSchema(w http.ResponseWriter, r *http.Request) { dec := json.NewDecoder(r.Body) dec.DisallowUnknownFields() schema := ingestSpec{} - err := dec.Decode(&schema) - if err != nil { - resp.write(w, err) - return - } - - // create index - indexName := schema.IndexName - resp.Name = indexName - var useKeys bool - switch schema.PrimaryKeyType { - case "string": - useKeys = true - case "uint": - useKeys = false - default: - resp.write(w, errors.New("Invalid primary key type")) - return - } - req := postIndexRequest{ - Options: pilosa.IndexOptions{ - Keys: useKeys, - TrackExistence: true, - }, - } - index, err := h.api.CreateIndex(r.Context(), indexName, req.Options) - if index != nil { - resp.CreatedAt = index.CreatedAt() - } - if err != nil { - if _, ok := errors.Cause(err).(pilosa.ConflictError); ok { - if index, _ = h.api.Index(r.Context(), indexName); index != nil { - resp.CreatedAt = index.CreatedAt() - } - if schema.IfNotExists { - // if the user did intend to solely create an index if not exists i.e. - // by setting it to true, then we should return an "OK" status code - // if the index already exists. We might want to upsert the fields - // in future for the sake of better UX, but for the time being, let's - // stick to the expected behaviour in SQL dbs by simply ignoring the - // rest of the schema specification - resp.write(w, nil) - } else { - // user indicated that if index already exists and they attempt - // to create index with same name, then it should error out - resp.write(w, err) - } - } else { - resp.write(w, err) - } - return - } - - // if any error occurs, rather than have a partially created index - // delete it entirely - var schemaErr error = nil + // if a key in cleanupIndexes points to a 0-length slice, the + // entire index should be cleaned; otherwise, only the named + // fields within that index should be cleaned. + cleanupIndexes := map[string][]string{} + var schemaErr error defer func() { + // we set schemaErr in any case where we need to do cleanup if schemaErr != nil { - _ = h.api.DeleteIndex(r.Context(), indexName) + for index, fields := range cleanupIndexes { + if len(fields) == 0 { + err := h.api.DeleteIndex(r.Context(), index) + if err != nil { + h.logger.Printf("deleting index %q after schema err: %v", index, err) + } + } else { + for _, field := range fields { + err := h.api.DeleteField(r.Context(), index, field) + if err != nil { + h.logger.Printf("deleting field %q from index %q after schema err: %v", field, index, err) + } + } + } + } } }() - - // create all the fields specified in the index - for _, fSpec := range schema.Fields { - opt := fieldSpecToFieldOption(fSpec) - schemaErr = opt.validate() - if schemaErr != nil { - resp.write(w, schemaErr) + for dec.More() { + err := dec.Decode(&schema) + if err != nil { + resp.write(w, err) return } - fos := fieldOptionsToFunctionalOpts(opt) - _, schemaErr = h.api.CreateField(r.Context(), indexName, fSpec.FieldName, fos...) - if schemaErr != nil { - if _, ok := schemaErr.(pilosa.BadRequestError); ok { - http.Error(w, schemaErr.Error(), http.StatusBadRequest) - } else if _, ok = errors.Cause(schemaErr).(pilosa.ConflictError); ok { - // TODO how to handle conflict if field already created - // current approach is to error out - http.Error(w, schemaErr.Error(), http.StatusConflict) - } else { - http.Error(w, schemaErr.Error(), http.StatusBadRequest) - } + index, fields, err := h.applyOneIngestSchema(r.Context(), &schema) + if err != nil { + // if a previous schema created things, clean them up... + schemaErr = err + resp.write(w, err) return } + // we only have one slot to report these, sorry. + resp.Name = index.Name() + resp.CreatedAt = index.CreatedAt() + cleanupIndexes[index.Name()] = fields + } + // if we got here, we have a cleanupIndexes which we want to return, + // so we want to do that *instead* of the successResponse we'd be + // using otherwise (ironically, to indicate an error) + var mapBody []byte + var err error + if mapBody, err = json.Marshal(cleanupIndexes); err != nil { + resp.write(w, err) + } + if _, err = w.Write(mapBody); err != nil { + h.logger.Printf("error trying to write response: %v", err) } - - resp.write(w, nil) } type postFieldRequest struct { diff --git a/http/handler_test.go b/http/handler_test.go index d91985bd6..bfea699d3 100644 --- a/http/handler_test.go +++ b/http/handler_test.go @@ -99,6 +99,7 @@ func TestIngestSchemaHandler(t *testing.T) { { "index-name": "example", "primary-key-type": "string", + "index-action": "create", "fields": [ { "field-name": "idset", diff --git a/ingest/codec.go b/ingest/codec.go index 108baada1..c966e3f1c 100644 --- a/ingest/codec.go +++ b/ingest/codec.go @@ -363,7 +363,7 @@ func (j *jsonFieldCodec) DecodeTimeValue(recID uint64, dataType jsonparser.Value if err != nil { return fmt.Errorf("parsing numeric timestamp: %w", err) } - j.currentOp.AddSignedPair(recID, i64-j.epoch) + j.currentOp.AddSignedPair(recID, i64) } return nil } @@ -464,6 +464,9 @@ func (codec *JSONCodec) ParseOperation(data []byte) (op *Operation, err error) { } return nil }) + if err != nil { + return nil, err + } if op.OpType == OpNone { return nil, fmt.Errorf("action not specified") } @@ -480,6 +483,11 @@ func (codec *JSONCodec) Parse(r io.Reader) (req *Request, err error) { return codec.ParseBytes(data) } +// ParseBytes reads a request from a slice of bytes. We need to use a +// byte slice because jsonparser's model is fundamentally built around +// being able to random-access the slice and return slices of it, so +// it can't really admit functional streaming. If we need streaming, the +// streaming needs to be at a higher level. func (codec *JSONCodec) ParseBytes(data []byte) (req *Request, err error) { var ops []*Operation var lastErr error @@ -541,6 +549,10 @@ func (codec *JSONCodec) ParseBytes(data []byte) (req *Request, err error) { continue } for field, fieldOp := range op.FieldOps { + if len(fieldOp.RecordIDs) == 0 { + delete(op.FieldOps, field) + continue + } if keyMap != nil { if err = fieldOp.TranslateKeys(keyMap); err != nil { return nil, fmt.Errorf("mapping record keys for op on %q: %w", field, err) diff --git a/ingest/codec_test.go b/ingest/codec_test.go index fe3479273..4490d3729 100644 --- a/ingest/codec_test.go +++ b/ingest/codec_test.go @@ -15,8 +15,11 @@ package ingest import ( + "fmt" "testing" "time" + + "github.com/molecula/featurebase/v2/shardwidth" ) func unusableSampleTranslator(keys ...string) (map[string]uint64, error) { @@ -30,12 +33,12 @@ func unusableSampleTranslator(keys ...string) (map[string]uint64, error) { func TestSimpleCodec(t *testing.T) { c, _ := NewJSONCodec(nil) _ = c.AddSetField("set", nil) - _ = c.AddSetField("setKeys", unusableSampleTranslator) + _ = c.AddSetField("setkeys", unusableSampleTranslator) _ = c.AddMutexField("mutex", nil) - _ = c.AddMutexField("mutexKeys", unusableSampleTranslator) + _ = c.AddMutexField("mutexkeys", unusableSampleTranslator) _ = c.AddTimeQuantumField("tq", nil) _ = c.AddIntField("int", nil) - _ = c.AddIntField("intKeys", unusableSampleTranslator) + _ = c.AddIntField("intkeys", unusableSampleTranslator) epoch, err := time.Parse("2006-01-02", "2020-01-01") if err != nil { t.Fatalf("can't parse sample epoch time: %v", err) @@ -43,7 +46,8 @@ func TestSimpleCodec(t *testing.T) { _ = c.AddTimestampField("ts", time.Millisecond, epoch.Unix()*1000) _ = c.AddDecimalField("dec", 2) _ = c.AddBoolField("bool") - sampleJson := []byte(` + var nextShard = uint64(1< 0 { - t.Logf(" clearRecordIDs: %d", op.ClearRecordIDs) - } - if len(op.ClearFields) > 0 { - t.Logf(" clearFields: %s", op.ClearFields) - } - for field, fieldOp := range op.FieldOps { - t.Logf(" field %q: %#v", field, fieldOp) - } - } - sharded, err := req.Shard() + // req.Dump(t.Logf) + sharded, err := req.ByShard() if err != nil { t.Errorf("sharding err: %v", err) } for shard, ops := range sharded.Ops { - t.Logf("shard %d:", shard) - for _, op := range ops { - t.Logf(" op: %q", op.OpType.String()) - if len(op.ClearRecordIDs) > 0 { - t.Logf(" clearRecordIDs: %d", op.ClearRecordIDs) + for i, op := range ops { + op.Sort() + for field, fop := range op.FieldOps { + sorter := fieldTypeSorts[req.FieldTypes[field]] + if sorter == nil { + sorter = (*FieldOperation).SortByRecords + } + sorter(fop) } - if len(op.ClearFields) > 0 { - t.Logf(" clearFields: %s", op.ClearFields) + var expectedOp *Operation + if i < len(expected[shard]) { + expectedOp = expected[shard][i] } - for field, fieldOp := range op.FieldOps { - t.Logf(" field %q: %#v", field, fieldOp) + if err = op.Compare(expectedOp); err != nil { + t.Errorf("shard %d, op %d: %v", shard, i, err) } } } diff --git a/ingest/op.go b/ingest/op.go index 3a1382e0e..b1d6255cd 100644 --- a/ingest/op.go +++ b/ingest/op.go @@ -17,6 +17,7 @@ package ingest import ( "fmt" "math/bits" + "sort" "github.com/molecula/featurebase/v2/shardwidth" ) @@ -98,6 +99,53 @@ type Operation struct { FieldOps map[string]*FieldOperation } +// Compare reports whether two operations seem to be the same. +func (got *Operation) Compare(expected *Operation) error { + if got == nil && expected == nil { + return nil + } + if got == nil { + return fmt.Errorf("expected %q op, got nil", expected.OpType) + } + if expected == nil { + return fmt.Errorf("expected no op, got %q", got.OpType) + } + if got.OpType != expected.OpType { + return fmt.Errorf("operation type mismatch: expected %q, got %q", expected.OpType, got.OpType) + } + if len(got.ClearRecordIDs) != len(expected.ClearRecordIDs) { + return fmt.Errorf("clear record counts differ: expected %d, got %d", len(expected.ClearRecordIDs), len(got.ClearRecordIDs)) + } + for i, v1 := range got.ClearRecordIDs { + v2 := expected.ClearRecordIDs[i] + if v1 != v2 { + return fmt.Errorf("clear record id %d differs: expected %d, got %d", i, v2, v1) + } + } + if len(got.ClearFields) != len(expected.ClearFields) { + return fmt.Errorf("clear field counts differ: expected %d (%q), got %d (%q)", len(expected.ClearFields), expected.ClearFields, len(got.ClearFields), got.ClearFields) + } + for i, v1 := range got.ClearFields { + v2 := expected.ClearFields[i] + if v1 != v2 { + return fmt.Errorf("clear field %d differs: expected %q, got %q", i, v2, v1) + } + } + for k, fo1 := range got.FieldOps { + fo2 := expected.FieldOps[k] + if err := fo1.Compare(fo2); err != nil { + return fmt.Errorf("field %q mismatch: %w", k, err) + } + } + for k := range expected.FieldOps { + _, ok := got.FieldOps[k] + if !ok { + return fmt.Errorf("expected op for field %q, but none found", k) + } + } + return nil +} + // FieldOperation is the specific set of changes to make to a given // field. // @@ -114,21 +162,22 @@ type FieldOperation struct { Signed []int64 } -// Sort sorts the values by record ID. It is not a stable sort. +// Sort sorts the clear record IDs and field list. func (o *Operation) Sort() { // I am aware that this is a crime, but it avoids rewriting // the code and justifies FieldOperation handling the "only record // IDs" case. f := FieldOperation{RecordIDs: o.ClearRecordIDs} f.SortByRecords() + sort.Strings(o.ClearFields) } type ShardedFieldOperation map[uint64]*FieldOperation -// Shard() divides the FieldOperation's values up into corresponding chunks +// ByShard() divides the FieldOperation's values up into corresponding chunks // based on the shards of record IDs. Does not further sort IDs within those // chunks. -func (f *FieldOperation) Shard() ShardedFieldOperation { +func (f *FieldOperation) ByShard() ShardedFieldOperation { if len(f.RecordIDs) == 0 { return nil } @@ -478,6 +527,57 @@ func (f *FieldOperation) AddStampedPair(rec uint64, value uint64, stamp int64) { f.Signed = append(f.Signed, stamp) } +// Compare returns a diagnostic if the field operations do not seem +// equivalent. +func (got *FieldOperation) Compare(expected *FieldOperation) error { + if got == nil { + if expected == nil { + return nil + } + if len(expected.RecordIDs) == 0 && len(expected.Values) == 0 && len(expected.Signed) == 0 { + return nil + } + return fmt.Errorf("expected field operation with %d records, got nil", len(expected.RecordIDs)) + } + if expected == nil { + if got == nil { + return nil + } + if len(got.RecordIDs) == 0 && len(got.Values) == 0 && len(got.Signed) == 0 { + return nil + } + return fmt.Errorf("expected empty field operation, got %d records", len(got.RecordIDs)) + } + if len(got.RecordIDs) != len(expected.RecordIDs) { + return fmt.Errorf("record counts differ: expected %d, got %d", len(expected.RecordIDs), len(got.RecordIDs)) + } + for i, v1 := range got.RecordIDs { + v2 := expected.RecordIDs[i] + if v1 != v2 { + return fmt.Errorf("record id %d differs: expected %d, got %d", i, v2, v1) + } + } + if len(got.Values) != len(expected.Values) { + return fmt.Errorf("value counts differ: expected %d, got %d", len(expected.Values), len(got.Values)) + } + for i, v1 := range got.Values { + v2 := expected.Values[i] + if v1 != v2 { + return fmt.Errorf("value %d differs: expected %d, got %d", i, v2, v1) + } + } + if len(got.Signed) != len(expected.Signed) { + return fmt.Errorf("signed value counts differ: expected %d, got %d", len(expected.Signed), len(got.Signed)) + } + for i, v1 := range got.Signed { + v2 := expected.Signed[i] + if v1 != v2 { + return fmt.Errorf("signed value %d differs: expected %d, got %d", i, v2, v1) + } + } + return nil +} + func translateUnsignedSlice(target []uint64, mapping []uint64) (err error) { oops := 0 for i, v := range target { @@ -544,8 +644,8 @@ type ShardedRequest struct { Ops map[uint64][]*Operation } -// Shard converts a request into the same request, only sharded. -func (r *Request) Shard() (*ShardedRequest, error) { +// ByShard converts a request into the same request, only sharded. +func (r *Request) ByShard() (*ShardedRequest, error) { if len(r.Ops) == 0 { return &ShardedRequest{Ops: nil}, nil } @@ -566,7 +666,7 @@ func (r *Request) Shard() (*ShardedRequest, error) { } } for field, fieldOp := range op.FieldOps { - sharded := fieldOp.Shard() + sharded := fieldOp.ByShard() sorter := fieldTypeSorts[r.FieldTypes[field]] if sorter == nil { sorter = (*FieldOperation).SortByRecords diff --git a/ingest/op_test.go b/ingest/op_test.go index 1988201f4..b0b196e07 100644 --- a/ingest/op_test.go +++ b/ingest/op_test.go @@ -124,7 +124,7 @@ var opShardingTestCases = []opShardingTestCase{ func TestOpSharding(t *testing.T) { for _, c := range opShardingTestCases { - sharded, err := c.input.Shard() + sharded, err := c.input.ByShard() if err != nil { t.Errorf("sharding: unexpected error %v", err) } diff --git a/ingest_test.go b/ingest_test.go new file mode 100644 index 000000000..649f5dd16 --- /dev/null +++ b/ingest_test.go @@ -0,0 +1,468 @@ +// Copyright 2021 Molecula Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package pilosa_test + +import ( + "bytes" + "context" + "encoding/json" + "io/ioutil" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + + pilosa "github.com/molecula/featurebase/v2" + "github.com/molecula/featurebase/v2/http" + "github.com/molecula/featurebase/v2/ingest" + "github.com/molecula/featurebase/v2/server" + "github.com/molecula/featurebase/v2/test" + "github.com/pkg/errors" +) + +// For ingest API testing, we want to do tests which have a known +// schema, no existing data before we start, and perform ingests and +// then do queries. +// +// A good starting point for this would be a fairly simple file +// divided into sections which are just the JSON text of the data +// we want to be working with, or the PQL queries we want to run, +// or their expected results. +// +// So, roughly like this: +// +// schema: +// { +// "index-name": "example", +// "primary-key-type": "string", +// "fields": [ +// { +// "field-name": "set", +// "field-type": "id", +// "field-options": { "cache-type": "none" } +// } +// ] +// } +// ingest: +// [ +// { +// "action": "set", +// "records": { +// "1": { +// "set": [ 2 ], +// } +// } +// } +// ] +// queries: +// Row(set=2): +// [1] +// +// Additionally, the names "schema-error" and "ingest-error" are taken to +// represent a schema, or data set, which is expected to produce an error. +// For instance: +// +// ingest-error: +// [ { "action": puppy } +// +// In this case, it would be considered a test failure if an ingest request +// did NOT fail. +// Lines starting with # + +// ingestSchemaPartial represents the only part of a schema we need to +// know about in order to undo its creation of a schema for use in a +// test case. +type ingestSchemaPartial struct { + IndexName string `json:"index-name"` +} + +type ingestActionKind int + +const ( + ingestActionNone = ingestActionKind(iota) + ingestActionSchema + ingestActionIngest + ingestActionSchemaError + ingestActionIngestError + ingestActionQueries +) + +var ingestActionKinds = map[string]ingestActionKind{ + "schema": ingestActionSchema, + "ingest": ingestActionIngest, + "schema-error": ingestActionSchemaError, + "ingest-error": ingestActionIngestError, + "queries": ingestActionQueries, +} +var ingestActionKindNames = map[ingestActionKind]string{} + +type testCaseAction struct { + kind ingestActionKind + comment []byte + lineStart int + lineEnd int + data []byte +} + +type liner struct { + data []byte + at int + start int + remaining []byte + line []byte +} + +func newLiner(data []byte) *liner { + return &liner{data: data, at: 0, remaining: data} +} + +func (l *liner) next() bool { + if len(l.remaining) == 0 { + return false + } + foundNL := true + nextNL := bytes.IndexByte(l.remaining, '\n') + if nextNL == -1 { + foundNL = false + nextNL = len(l.remaining) + } + l.line = l.remaining[:nextNL] + // move past the newline we found, if we found one + if foundNL { + nextNL++ + } + l.start, l.at = l.at, l.at+nextNL + l.remaining = l.data[l.at:] + return true +} + +func (l *liner) text() (line []byte, start int, end int) { + return l.line, l.start, l.at +} + +// parseExpectedResults handles something that looks like +// [1, 2, 3] or ["a", "b", "c"]. It does not handle things like +// quotes within strings, etcetera. +func parseExpectedResults(data []byte) (ints []uint64, keys []string, err error) { + if len(data) < 2 || data[0] != '[' || data[len(data)-1] != ']' { + return nil, nil, errors.New("expecting [] results") + } + words := bytes.Split(data[1:len(data)-1], []byte{','}) + for _, word := range words { + word = bytes.TrimSpace(word) + if len(word) == 0 { + return nil, nil, errors.New("found empty word expecting result") + } + if word[0] == '"' { + keys = append(keys, string(word[1:len(word)-1])) + continue + } + v, err := strconv.ParseInt(string(word), 10, 64) + if err != nil { + return nil, nil, err + } + ints = append(ints, uint64(v)) + } + if len(ints) > 0 && len(keys) > 0 { + return nil, nil, errors.New("mixed integers and strings are invalid") + } + return ints, keys, err +} + +func testQueries(t *testing.T, ctx context.Context, cmd *test.Command, index string, action testCaseAction) { + qcx := cmd.API.Txf().NewQcx() + defer func() { + if err := qcx.Finish(); err != nil { + t.Fatalf("finishing qcx: %v", err) + } + }() + l := newLiner(action.data) + for l.next() { + query, _, _ := l.text() + if !l.next() { + t.Fatalf("processing query list: no expected after %q", query) + } + expected, _, _ := l.text() + ints, keys, err := parseExpectedResults(expected) + if err != nil { + t.Fatalf("processing query list: invalid expected results %q", expected) + } + t.Logf("expecting %q -> %s", query, expected) + res, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: string(query)}) + if err != nil { + t.Errorf("query: %v", err) + } + if len(res.Results) != 1 { + t.Fatalf("expected one result per query, got %d results", len(res.Results)) + } + var row *pilosa.Row + var ok bool + if row, ok = res.Results[0].(*pilosa.Row); !ok { + t.Fatalf("expected results to be a row") + } + if ints != nil { + cols := row.Columns() + if len(cols) != len(ints) { + t.Fatalf("wrong number of values, expected %d, got %d", len(ints), len(cols)) + } + for i := range cols { + if ints[i] != cols[i] { + t.Fatalf("result %d: expected %d, got %d", i, ints[i], cols[i]) + } + } + } + // key return value is unpredictable, so... + if keys != nil { + seen := make(map[string]struct{}) + for _, k := range keys { + seen[k] = struct{}{} + } + for _, k := range row.Keys { + if _, ok := seen[k]; !ok { + t.Fatalf("unexpected result key %q", k) + } + delete(seen, k) + } + for k := range seen { + t.Fatalf("expected result to contain %q, but did not get it", k) + } + } + } +} + +// testOneIngestTestcase runs a set of actions, then cleans up after itself +func testOneIngestTestcase(t *testing.T, ctx context.Context, cmd *test.Command, tcpath string) { + data, err := ioutil.ReadFile(tcpath) + if err != nil { + t.Fatalf("reading %q: %v", tcpath, err) + } + var actions []testCaseAction + var action testCaseAction + l := newLiner(data) + var line []byte + var start, lineStart, lineEnd int + lineCount := 0 + for l.next() { + line, lineStart, lineEnd = l.text() + lineCount++ + if colon := bytes.IndexByte(line, ':'); colon != -1 { + if kind, ok := ingestActionKinds[string(line[:colon])]; ok { + if action.kind != ingestActionNone { + action.data = data[start:lineStart] + actions = append(actions, action) + action.lineEnd = lineCount - 1 + } else { + if lineStart != 0 { + t.Logf("warning: %d bytes with no action type before first action", lineStart) + } + } + start = lineEnd + action.data = nil + action.kind = kind + if line[len(line)-1] == ':' { + action.comment = line[:len(line)-1] + } else { + action.comment = line + } + action.lineStart = lineCount + } + } + } + if action.kind != ingestActionNone { + action.lineEnd = lineCount - 1 + action.data = data[start:] + actions = append(actions, action) + } + var mostRecentIndex string + seenIndexes := map[string]struct{}{} + + cli := cmd.Client() + created := map[string][]string{} + defer func() { + t.Logf("deleting created indexes/fields:") + for k, v := range created { + if len(v) == 0 { + t.Logf(" index: %q", k) + if err := cmd.API.DeleteIndex(ctx, k); err != nil { + t.Errorf("deleting index %q: %v", k, err) + } + } else { + t.Logf(" fields in %q: %q", k, v) + for _, field := range v { + if err := cmd.API.DeleteField(ctx, k, field); err != nil { + t.Errorf("deleting field %q from %q: %v", field, k, err) + } + } + } + } + }() + + noticeCreation := func(newlyCreated map[string][]string) { + for k, v := range newlyCreated { + if len(v) == 0 { + if existing, ok := created[k]; ok { + if len(existing) > 0 { + t.Fatalf("creation reports index %q newly created, but we created fields %q in it previously", + k, existing) + } + } + // create an empty list, indicating that the whole index is + // believed nil + created[k] = nil + continue + } + if existing, ok := created[k]; ok { + if len(existing) == 0 { + // we'll delete this index anyway, don't need to delete fields in it + continue + } + created[k] = append(existing, v...) + continue + } + created[k] = v + } + } + + for _, action := range actions { + t.Logf("%s, lines %d-%d", action.comment, action.lineStart, action.lineEnd) + switch action.kind { + case ingestActionSchema: + var scratch ingestSchemaPartial + err = json.Unmarshal(action.data, &scratch) + if err != nil { + t.Fatalf("couldn't parse schema data: %v", err) + } + if scratch.IndexName == "" { + t.Fatalf("test case must provide an index name") + } + // stash the string from the schema, because we + // might need it later + mostRecentIndex = scratch.IndexName + seenIndexes[mostRecentIndex] = struct{}{} + var newlyCreated map[string][]string + newlyCreated, err = cli.IngestSchema(ctx, nil, action.data) + if err != nil { + t.Fatalf("executing schema: %v", err) + } + noticeCreation(newlyCreated) + case ingestActionSchemaError: + var scratch ingestSchemaPartial + err = json.Unmarshal(action.data, &scratch) + if err != nil { + t.Logf("got expected error from schema: %v", err) + break + } + var newlyCreated map[string][]string + newlyCreated, err = cli.IngestSchema(ctx, nil, action.data) + if err != nil { + t.Logf("got expected error from schema: %v", err) + break + } + noticeCreation(newlyCreated) + t.Fatalf("expected error from schema, didn't get it") + case ingestActionIngest: + func() { + qcx := cmd.API.Txf().NewQcx() + var err error + defer func() { + if err == nil { + qcx.Abort() + return + } + if err := qcx.Finish(); err != nil { + t.Fatalf("finishing qcx: %v", err) + } + }() + err = cmd.API.IngestOperations(ctx, qcx, mostRecentIndex, bytes.NewBuffer(action.data)) + if err != nil { + t.Fatalf("importing data: %v", err) + } + }() + case ingestActionIngestError: + func() { + qcx := cmd.API.Txf().NewQcx() + var err error + defer func() { + if err == nil { + qcx.Abort() + return + } + if err := qcx.Finish(); err != nil { + t.Fatalf("finishing qcx: %v", err) + } + }() + err = cmd.API.IngestOperations(ctx, qcx, mostRecentIndex, bytes.NewBuffer(action.data)) + if err != nil { + t.Logf("got expected error from ingest: %v", err) + return + } + t.Fatalf("expected error from ingest, didn't get it") + }() + case ingestActionQueries: + testQueries(t, ctx, cmd, mostRecentIndex, action) + } + } + +} + +// TestIngestTestcases reads sample test cases from a test data directory +// and evaluates them. +func TestIngestTestcases(t *testing.T) { + _ = &ingest.Operation{} + var testcases []string + if len(ingestActionKindNames) == 0 { + for k, v := range ingestActionKinds { + ingestActionKindNames[v] = k + } + } + err := filepath.Walk("ingest_testdata", func(path string, info os.FileInfo, err error) error { + if err != nil { + return err + } + if info.IsDir() { + return nil + } + if strings.HasSuffix(path, ".tc") { + testcases = append(testcases, path) + } + return nil + }) + if err != nil { + t.Fatalf("looking for test cases: %v", err) + } + if len(testcases) == 0 { + t.Fatalf("no ingest test cases found") + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + c := test.MustRunCluster(t, 1, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID("node0"), + pilosa.OptServerClusterHasher(&offsetModHasher{}), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + )}, + ) + defer c.Close() + + coord := c.GetPrimary() + + for _, tc := range testcases { + t.Run(strings.TrimSuffix(tc, ".tc"), func(t *testing.T) { + testOneIngestTestcase(t, ctx, coord, tc) + }) + } +} diff --git a/ingest_testdata/bool.tc b/ingest_testdata/bool.tc new file mode 100644 index 000000000..a5957cb15 --- /dev/null +++ b/ingest_testdata/bool.tc @@ -0,0 +1,17 @@ +schema: +{ + "index-name": "example", + "index-action": "create", + "primary-key-type": "uint", + "fields": [ + { + "field-name": "tf", + "field-type": "bool" + } + ] +} +ingest: +[{"action": "write", "records": {"2": { "tf": true }}}] +queries: +Row(tf=true) +[2] diff --git a/ingest_testdata/expect_errors.tc b/ingest_testdata/expect_errors.tc new file mode 100644 index 000000000..707bfcdba --- /dev/null +++ b/ingest_testdata/expect_errors.tc @@ -0,0 +1,101 @@ +schema-error: fail to create field +{ + "index-name": "examplekeys", + "index-action": "create", + "primary-key-type": "string", + "fields": [ + { + "field-name": "cookie", + "field-type": "id", + "field-options": { "cache-type": "none" } + }, + { + "field-name": "set", + "field-type": "id", + "field-options": { "cache-type": "nun" } + } + ] +} +schema: confirm index was not created because field failed +{ + "index-name": "examplekeys", + "index-action": "create", + "primary-key-type": "string", + "fields": [ + { + "field-name": "set", + "field-type": "id", + "field-options": { "cache-type": "none" } + } + ] +} +schema-error: can't recreate index +{ + "index-name": "examplekeys", + "index-action": "create", + "primary-key-type": "string", + "fields": [ + { + "field-name": "newfield", + "field-type": "id", + "field-options": { "cache-type": "none" } + } + ] +} +schema: can add field to existing index +{ + "index-name": "examplekeys", + "index-action": "ensure", + "primary-key-type": "string", + "fields": [ + { + "field-name": "newfield", + "field-type": "id", + "field-options": { "cache-type": "none" } + } + ] +} +schema-error: second field failing in existing index +{ + "index-name": "examplekeys", + "index-action": "ensure", + "primary-key-type": "string", + "fields": [ + { + "field-name": "addokay", + "field-type": "id", + "field-options": { "cache-type": "none" } + }, + { + "field-name": "addfail", + "field-type": "id", + "field-options": { "cache-type": "nun" } + } + ] +} +schema: verify that fields we think exist do exist +{ + "index-name": "examplekeys", + "index-action": "require", + "primary-key-type": "string", + "fields": [ + { + "field-name": "newfield", + "field-type": "id", + "field-options": { "cache-type": "none" } + } + ] +} +schema-error: successful field deleted anyway because second field failed +{ + "index-name": "examplekeys", + "index-action": "require", + "primary-key-type": "string", + "fields": [ + { + "field-name": "addokay", + "field-type": "id", + "field-options": { "cache-type": "none" } + } + ] +} diff --git a/ingest_testdata/keyed.tc b/ingest_testdata/keyed.tc new file mode 100644 index 000000000..8f7d5c4b6 --- /dev/null +++ b/ingest_testdata/keyed.tc @@ -0,0 +1,32 @@ +schema: +{ + "index-name": "examplekeys", + "index-action": "create", + "primary-key-type": "string", + "fields": [ + { + "field-name": "set", + "field-type": "id", + "field-options": { "cache-type": "none" } + } + ] +} +ingest: +[ + { + "action": "set", + "records": { + "a": { + "set": [ 2 ], + }, + "b": { + "set": 3, + } + } + } +] +queries: +Row(set=2) +["a"] +Union(Row(set=3),Row(set=2)) +["a","b"] diff --git a/ingest_testdata/sample.tc b/ingest_testdata/sample.tc new file mode 100644 index 000000000..6ddff22a4 --- /dev/null +++ b/ingest_testdata/sample.tc @@ -0,0 +1,89 @@ +schema: +{ + "index-name": "example", + "index-action": "create", + "primary-key-type": "uint", + "fields": [ + { + "field-name": "set", + "field-type": "id", + "field-options": { "cache-type": "none" } + } + ] +} +ingest: +[ + { + "action": "set", + "records": { + "1": { + "set": [ 2 ], + }, + "2": { + "set": 3, + } + } + } +] +queries: +Row(set=2) +[1] +Union(Row(set=3),Row(set=2)) +[1,2] +schema-error: +{ + "index-name": "example", + "primary-key-type": "uint", + "index-action": "require", + "fields": [ + { + "field-name": "setkey", + "field-type": "string", + "field-options": { "cache-type": "none" } + } + ] +} +schema: +{ + "index-name": "example", + "primary-key-type": "uint", + "index-action": "ensure", + "fields": [ + { + "field-name": "setkey", + "field-type": "string", + "field-options": { "cache-type": "none" } + } + ] +} +ingest: +[ + { + "action": "set", + "records": { + "1": { + "setkey": [ "a" ], + }, + "2": { + "setkey": "b", + } + } + } +] +queries: +Row(setkey="a") +[1] +ingest-error: +[ + { + "action": "setkeys", + "records": { + "a": { + "setkey": [ "a" ], + }, + "b": { + "setkey": "b", + } + } + } +] diff --git a/tracker.go b/tracker.go index 3de1ca857..280e6c366 100644 --- a/tracker.go +++ b/tracker.go @@ -34,7 +34,7 @@ type PastQueryStatus struct { Node string `json:"nodeID"` Index string `json:"index"` Start time.Time `json:"start"` - Runtime time.Duration `json:"runtime"` // deprecated + Runtime time.Duration `json:"runtime"` // deprecated RuntimeNs time.Duration `json:"runtimeNanoseconds"` }