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

ingest and ingest/codec testing work

Browse source 
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.
This commit is contained in:
Seebs 2021-08-02 16:33:08 -05:00 • committed by Seebs
parent 3185fe4181
commit bb1d52a385
16 changed files with 1254 additions and 150 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
api.go
View file

@ -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)

28
api_test.go
View file

@ -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++ {

8
fragment.go
View file

@ -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

45
http/client.go
View file

@ -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

276
http/handler.go
View file

@ -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 {

1
http/handler_test.go
View file

@ -99,6 +99,7 @@ func TestIngestSchemaHandler(t *testing.T) {
{
"index-name": "example",
"primary-key-type": "string",
"index-action": "create",
"fields": [
{
"field-name": "idset",

14
ingest/codec.go
View file

@ -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)

195
ingest/codec_test.go
View file

@ -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<<shardwidth.Exponent) + 5
sampleJson := []byte(fmt.Sprintf(`
[
{
"action": "set",
@ -57,17 +61,17 @@ func TestSimpleCodec(t *testing.T) {
"dec": 1.02,
"int": 3,
"mutex": 4,
"mutexKeys": "key-a",
"intKeys": "key-a"
"mutexkeys": "key-a",
"intkeys": "key-a"
},
"1234567": {
"%d": {
"set": [ 3 ],
"bool": true
},
"1": {
"set": [ 2 ],
"bool": false,
"setKeys": [
"setkeys": [
"key-a",
"key-b"
],
@ -84,16 +88,16 @@ func TestSimpleCodec(t *testing.T) {
"action": "write",
"records": {
"3": {
"set": [ 2 ],
"set": 2,
"mutex": 4,
"mutexKeys": "key-a",
"mutexkeys": "key-a",
"tq": {
"time": "2006-01-02T15:04:05.999999999Z",
"values": [ 6 ]
"values": 6
},
"ts": "2020-01-01T00:01:00.000000000Z",
"int": 3,
"intKeys": "key-a"
"intkeys": "key-a"
},
"4": {
"set": [ 3 ],
@ -101,7 +105,7 @@ func TestSimpleCodec(t *testing.T) {
},
"5": {
"set": [ 2 ],
"setKeys": [ "key-a", "key-b" ],
"setkeys": [ "key-a", "key-b" ],
"tq": { "values": [ 3, 4 ] }
}
}
@ -115,45 +119,160 @@ func TestSimpleCodec(t *testing.T) {
"records": {
"2": {
"mutex": 5,
"mutexKeys": "key-b"
"mutexkeys": "key-b"
}
}
}
]
`)
`, nextShard))
var expected = map[uint64][]*Operation{
0: {
{
OpType: OpSet,
FieldOps: map[string]*FieldOperation{
"mutex": {
RecordIDs: []uint64{2},
Values: []uint64{4},
},
"set": {
RecordIDs: []uint64{1, 2},
Values: []uint64{2, 2},
},
"tq": {
RecordIDs: []uint64{1, 1, 2},
Values: []uint64{3, 4, 6},
Signed: []int64{0, 0, 1136214245999999999},
},
"mutexkeys": {
RecordIDs: []uint64{2},
Values: []uint64{0},
},
"int": {
RecordIDs: []uint64{2},
Signed: []int64{3},
},
"intkeys": {
RecordIDs: []uint64{2},
Signed: []int64{0},
},
"setkeys": {
RecordIDs: []uint64{1, 1},
Values: []uint64{0, 13},
},
"dec": {
RecordIDs: []uint64{2},
Signed: []int64{102},
},
"bool": {
RecordIDs: []uint64{1},
Values: []uint64{0},
},
},
},
{
OpType: OpClear,
ClearRecordIDs: []uint64{5, 6, 7},
ClearFields: []string{"tq"},
},
{
OpType: OpWrite,
ClearRecordIDs: []uint64{3, 4, 5},
ClearFields: []string{"int", "intkeys", "mutex", "mutexkeys", "set", "setkeys", "tq", "ts"},
FieldOps: map[string]*FieldOperation{
"mutex": {
RecordIDs: []uint64{3},
Values: []uint64{4},
},
"set": {
RecordIDs: []uint64{3, 5, 4},
Values: []uint64{2, 2, 3},
},
"tq": {
RecordIDs: []uint64{5, 5, 3},
Values: []uint64{3, 4, 6},
Signed: []int64{0, 0, 1136214245999999999},
},
"mutexkeys": {
RecordIDs: []uint64{3},
Values: []uint64{0},
},
"int": {
RecordIDs: []uint64{3},
Signed: []int64{3},
},
"intkeys": {
RecordIDs: []uint64{3},
Signed: []int64{0},
},
"setkeys": {
RecordIDs: []uint64{5, 5},
Values: []uint64{0, 13},
},
"ts": {
RecordIDs: []uint64{3, 4},
Signed: []int64{60000, 1577836860000},
},
},
},
{
OpType: OpDelete,
ClearRecordIDs: []uint64{9},
},
{
OpType: OpSet,
FieldOps: map[string]*FieldOperation{
"mutex": {
RecordIDs: []uint64{2},
Values: []uint64{5},
},
"mutexkeys": {
RecordIDs: []uint64{2},
Values: []uint64{13},
},
},
},
},
1: {
{
OpType: OpSet,
FieldOps: map[string]*FieldOperation{
"set": {
RecordIDs: []uint64{nextShard},
Values: []uint64{3},
},
"bool": {
RecordIDs: []uint64{nextShard},
Values: []uint64{1},
},
},
},
},
}
req, err := c.ParseBytes(sampleJson)
if err != nil {
t.Fatalf("parsing sample buffer: %v", err)
}
t.Logf("req: %#v", req)
for _, op := range req.Ops {
t.Logf("op: %#v", op)
if len(op.ClearRecordIDs) > 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)
}
}
}

112
ingest/op.go
View file

@ -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

2
ingest/op_test.go
View file

@ -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)
}

468
ingest_test.go Normal file
View file

@ -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)
})
}
}

17
ingest_testdata/bool.tc Normal file
View file

@ -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]

101
ingest_testdata/expect_errors.tc Normal file
View file

@ -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" }
}
]
}

32
ingest_testdata/keyed.tc Normal file
View file

@ -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"]

89
ingest_testdata/sample.tc Normal file
View file

@ -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",
}
}
}
]

2
tracker.go
View file

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