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"` }