diff --git a/api.go b/api.go index bbdb8c6b4..aa7cdc79a 100644 --- a/api.go +++ b/api.go @@ -35,6 +35,7 @@ import ( "time" "github.com/molecula/featurebase/v2/disco" + "github.com/molecula/featurebase/v2/ingest" "github.com/molecula/featurebase/v2/pql" "github.com/molecula/featurebase/v2/roaring" "github.com/molecula/featurebase/v2/stats" @@ -1393,6 +1394,7 @@ type ImportOptions struct { Clear bool IgnoreKeyCheck bool Presorted bool + fullySorted bool // format-aware sorting, internal use only please. // test Tx atomicity if > 0 SimPowerLossAfter int @@ -1426,6 +1428,13 @@ func OptImportOptionsPresorted(b bool) ImportOption { } } +func optImportOptionsFullySorted(b bool) ImportOption { + return func(o *ImportOptions) error { + o.fullySorted = b + return nil + } +} + var ErrAborted = fmt.Errorf("error: update was aborted") func (api *API) ImportAtomicRecord(ctx context.Context, qcx *Qcx, req *AtomicRecord, opts ...ImportOption) error { @@ -1815,6 +1824,180 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu return nil } +func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string, stream io.Reader) error { + span, _ := tracing.StartSpanFromContext(ctx, "API.IngestOperations") + defer span.Finish() + + if err := api.validate(apiIngestOperations); err != nil { + return errors.Wrap(err, "validating api method") + } + + // Find the Index. + index := api.holder.Index(indexName) + if index == nil { + api.server.logger.Errorf("ingest: no such index %q", indexName) + return newNotFoundError(ErrIndexNotFound, indexName) + } + fields := index.Fields() + var lookup ingest.KeyLookupFunc + if index.Keys() { + lookup = func(keys ...string) (map[string]uint64, error) { + return api.cluster.createIndexKeys(ctx, indexName, keys...) + } + } + codec, err := ingest.NewJSONCodec(lookup) + if err != nil { + return errors.Wrap(err, "creating JSON codec") + } + knownFields := map[string]*Field{} + for _, field := range fields { + var lookup ingest.KeyLookupFunc + if field.usesKeys { + lookup = field.translateStore.CreateKeys + } + knownFields[field.name] = field + switch field.Type() { + case "set": + if err = codec.AddSetField(field.name, lookup); err != nil { + return fmt.Errorf("adding set field to codec: %w", err) + } + case "time": + if err = codec.AddTimeQuantumField(field.name, lookup); err != nil { + return fmt.Errorf("adding time quantum field to codec: %w", err) + } + case "mutex": + if err = codec.AddMutexField(field.name, lookup); err != nil { + return fmt.Errorf("adding mutex field to codec: %w", err) + } + case "bool": + if err = codec.AddBoolField(field.name); err != nil { + 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 int field to codec: %w", err) + } + case "decimal": + if err = codec.AddDecimalField(field.name, field.options.Scale); err != nil { + 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 timestamp field to codec: %w", err) + } + default: + return fmt.Errorf("unhandled field type %q", field.Type()) + } + } + req, err := codec.Parse(stream) + if err != nil { + return errors.Wrap(err, "parsing input data") + } + sharded, err := req.ByShard() + if err != nil { + return errors.Wrap(err, "sharding input data") + } + eg, ctx := errgroup.WithContext(ctx) + for shard, ops := range sharded.Ops { + // loop variable shadow capture is the go equivalent of man door hook hand + shard, ops := shard, ops + eg.Go(func() error { + return api.applyOperations(ctx, qcx, index, shard, knownFields, ops) + }) + } + return eg.Wait() +} + +// applyOperations applies a set of operations to one specific shard. +func (api *API) applyOperations(ctx context.Context, qcx *Qcx, index *Index, shard uint64, fields map[string]*Field, ops []*ingest.Operation) error { + // For each operation, we may have a set of records/fields to clear, and then + // also a set of fields to set/remove specific bits in. + opts := &ImportOptions{Presorted: true, IgnoreKeyCheck: true, fullySorted: true} + funcOpts := []ImportOption{OptImportOptionsIgnoreKeyCheck(true), OptImportOptionsPresorted(true), optImportOptionsFullySorted(true), OptImportOptionsClear(true)} + for _, op := range ops { + // ClearRecordIDs should exist only for delete, clear, and write. For clear and write, + // we'll have a list of fields, for delete, it should be all the fields. + if len(op.ClearRecordIDs) > 0 { + // anonymous func lets us defer a finisher from any of the inner error returns + err := func() (e0 error) { + // WARNING: Depends on GetTx being per-shard/index, not per-field. + tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: index, Shard: shard}) + if err != nil { + return fmt.Errorf("getting Tx: %w", err) + } + defer finisher(&e0) + // For a delete, we don't look at the fields the codec was defined with, + // We delete from the existence field unconditionally and other fields + // if we know they exist. + if op.OpType == ingest.OpDelete { + err = clearExistenceColumns(qcx, index, op.ClearRecordIDs, shard) + if err != nil { + return fmt.Errorf("clearing existence columns: %w", err) + } + for name, field := range fields { + if err = field.ClearBits(tx, shard, op.ClearRecordIDs...); err != nil { + return fmt.Errorf("clearing field %q: %w", name, err) + } + } + return nil + } + // clear things that we need to wipe out, whether it's because + // this is a Clear op, or because it's a write op that + // specifies clears for the fields it's going to write to. + if len(op.ClearFields) > 0 { + for _, fieldName := range op.ClearFields { + field, ok := fields[fieldName] + if !ok { + return fmt.Errorf("can't find a field named %q", fieldName) + } + if err = field.ClearBits(tx, shard, op.ClearRecordIDs...); err != nil { + return fmt.Errorf("clearing record IDs: %w", err) + } + } + } + return nil + }() + if err != nil { + return err + } + } + opts.Clear = (op.OpType == ingest.OpRemove) + // reslice this rather than regenerating it. i'm so efficient. + if opts.Clear { + funcOpts = funcOpts[:4] + } else { + funcOpts = funcOpts[:3] + } + // for "set" and "write" ops, we'll be setting bits, for + // "remove" ops we'll be clearing them, and for "clear" ops + // there shouldn't be anything here. + for fieldName, fieldOp := range op.FieldOps { + field, ok := fields[fieldName] + if !ok { + return fmt.Errorf("can't find a field named %q", fieldName) + } + var err error + err = importExistenceColumns(qcx, index, fieldOp.RecordIDs, shard) + if err != nil { + return errors.Wrap(err, "importing existence columns") + } + switch field.Type() { + 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) + default: + err = fmt.Errorf("unhandled field type %q", field.Type()) + } + if err != nil { + return err + } + } + } + return nil +} + func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uint64) error { ef := index.existenceField() if ef == nil { @@ -1831,6 +2014,22 @@ func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard ui return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard) } +func clearExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uint64) error { + ef := index.existenceField() + if ef == nil { + return nil + } + + existenceRowIDs := make([]uint64, len(columnIDs)) + // If we don't gratuitously hand-duplicate things in field.Import, + // the fact that fragment.bulkImport rewrites its row and column + // lists can burn us if we don't make a copy before doing the + // existence field write. + columnCopy := make([]uint64, len(columnIDs)) + copy(columnCopy, columnIDs) + return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, OptImportOptionsClear(true)) +} + // ShardDistribution returns an object representing the distribution of shards // across nodes for each index, distinguishing between primary and replica. // The structure of this information is [indexName][nodeID][primaryOrReplica][]uint64. @@ -2513,6 +2712,7 @@ const ( apiIDCommit apiIDReset apiPartitionNodes + apiIngestOperations ) var methodsCommon = map[apiMethod]struct{}{ @@ -2580,4 +2780,5 @@ var methodsNormal = map[apiMethod]struct{}{ apiIDCommit: {}, apiIDReset: {}, apiPartitionNodes: {}, + apiIngestOperations: {}, } diff --git a/api_test.go b/api_test.go index 201a7c89a..7e34f5be4 100644 --- a/api_test.go +++ b/api_test.go @@ -15,11 +15,12 @@ package pilosa_test import ( + "bytes" "context" - "crypto/rand" "errors" "fmt" "math" + "math/rand" "reflect" "strings" "testing" @@ -411,6 +412,179 @@ func TestAPI_ImportValue(t *testing.T) { }) } +func TestAPI_Ingest(t *testing.T) { + 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() + // m0 := c.GetNode(0) + // m1 := c.GetNode(1) + // m2 := c.GetNode(2) + + index := "ingest" + setField := "set" + timeField := "tq" + + _, err := coord.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: false}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + _, err = coord.API.CreateField(ctx, index, setField, pilosa.OptFieldTypeSet("none", 0)) + if err != nil { + t.Fatalf("creating field: %v", err) + } + _, err = coord.API.CreateField(ctx, index, timeField, pilosa.OptFieldTypeTime("YMD")) + if err != nil { + t.Fatalf("creating field: %v", err) + } + sampleJson := []byte(` +[ + { + "action": "set", + "records": { + "2": { + "set": [2], + "tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [6] } + }, + "5": { "set": [3] }, + "8": { "set": [3] }, + "1": { + "set": [2], + "tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [3, 4] } + }, + "4": { "set": [3, 7] } + } + }, + { + "action": "clear", + "record_ids": [ 5, 6, 7 ], + "fields": [ "tq", "set" ] + }, + { + "action": "write", + "records": { + "8": { "tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [3, 4] } }, + "9": { "set": [7, 3] } + } + }, + { + "action": "delete", + "record_ids": [ 9 ] + } +] +`) + // just for set row 3: + // first operation should set it for 4, 5, and 8. + // clear operation should clear it for 5, 6, and 7, leaving it still set for 4 and 8. + // the write operation should clear set for record 8, even though record 8 doesn't + // contain that field in that op, because set is present in record 9, which also + // gets row 3 set. but then we delete 9. + // so after all that we expect Row(set=3) to be 4... + sampleBuf := bytes.NewBuffer(sampleJson) + qcx := coord.API.Txf().NewQcx() + defer func() { + if err := qcx.Finish(); err != nil { + t.Fatalf("finishing qcx: %v", err) + } + }() + err = coord.API.IngestOperations(ctx, qcx, index, sampleBuf) + if err != nil { + t.Fatalf("importing data: %v", err) + } + query := "Row(set=3)" + res, err := coord.API.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query}) + if err != nil { + t.Errorf("query: %v", err) + } + r := res.Results[0].(*pilosa.Row).Columns() + if len(r) != 1 || r[0] != 4 { + t.Fatalf("expected row with 4 set, got %d", r) + } +} + +// 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++ { + 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, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID("node0"), + pilosa.OptServerClusterHasher(&offsetModHasher{}), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + )}, + ) + defer c.Close() + + coord := c.GetPrimary() + m0 := c.GetNode(0) + // m1 := c.GetNode(1) + // m2 := c.GetNode(2) + + 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) + } + _, err = coord.API.CreateField(ctx, index, setField, pilosa.OptFieldTypeSet("none", 0)) + 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++ { + qcx := m0.API.Txf().NewQcx() + defer qcx.Abort() + err = coord.API.IngestOperations(ctx, qcx, index, bytes.NewBuffer(data)) + if err != nil { + b.Fatalf("ingest: %v", err) + } + err = qcx.Finish() + if err != nil { + b.Fatalf("finish: %v", err) + } + } +} + // offsetModHasher represents a simple, mod-based hashing offset by 1. type offsetModHasher struct{} diff --git a/field.go b/field.go index 5f147d735..a1c495d68 100644 --- a/field.go +++ b/field.go @@ -1178,6 +1178,29 @@ func (f *Field) ClearBit(tx Tx, rowID, colID uint64) (changed bool, err error) { return changed, nil } +// ClearBits clears all bits corresponding to the given record IDs in standard +// or BSI views. It does not delete bits from time quantum views. +func (f *Field) ClearBits(tx Tx, shard uint64, recordIDs ...uint64) error { + bsig := f.bsiGroup(f.name) + var v *view + if bsig != nil { + // looks like we're a BSI field? + v = f.view(viewBSIGroupPrefix + f.name) + } else { + v = f.view(viewStandard) + } + // it's fine if we never actually created the view, that means the + // bits are all clear! + if v == nil { + return nil + } + frag := v.Fragment(shard) + if frag == nil { + return nil + } + return frag.ClearRecords(tx, recordIDs) +} + func groupCompare(a, b string, offset int) (lt, eq bool) { if len(a) > offset { a = a[:offset] diff --git a/fragment.go b/fragment.go index 8b7f55f46..844cc41df 100644 --- a/fragment.go +++ b/fragment.go @@ -1741,7 +1741,7 @@ func (f *fragment) pos(rowID, columnID uint64) (uint64, error) { // Return an error if the column ID is out of the range of the fragment's shard. minColumnID := f.shard * ShardWidth if columnID < minColumnID || columnID >= minColumnID+ShardWidth { - return 0, errors.Errorf("column:%d out of bounds", columnID) + return 0, errors.Errorf("column:%d out of bounds for shard %d", columnID, f.shard) } return pos(rowID, columnID), nil } @@ -2210,7 +2210,7 @@ func (f *fragment) bulkImport(tx Tx, rowIDs, columnIDs []uint64, options *Import } if f.mutexVector != nil && !options.Clear { - return f.bulkImportMutex(tx, rowIDs, columnIDs) + return f.bulkImportMutex(tx, rowIDs, columnIDs, options) } return f.bulkImportStandard(tx, rowIDs, columnIDs, options) } @@ -2253,8 +2253,12 @@ func (f *fragment) bulkImportStandard(tx Tx, rowIDs, columnIDs []uint64, options rowSet := make(map[uint64]struct{}) lastRowID := uint64(1 << 63) + // It's possible for the ingest API to have already sorted things in + // the row-first order we want for this import. + if !options.fullySorted { + sort.Sort(rowColumnSet{r: rowIDs, c: columnIDs}) + } // replace columnIDs with calculated positions to avoid allocation. - sort.Sort(rowColumnSet{r: rowIDs, c: columnIDs}) prevRow, prevCol := ^uint64(0), ^uint64(0) next := 0 for i := 0; i < len(columnIDs); i++ { @@ -2525,14 +2529,20 @@ func sliceDifference(original, remove []uint64) []uint64 { // mutex restrictions. Because the mutex requirements must be checked // against storage, this method must acquire a write lock on the fragment // during the entire process, and it handles every bit independently. -func (f *fragment) bulkImportMutex(tx Tx, rowIDs, columnIDs []uint64) error { +func (f *fragment) bulkImportMutex(tx Tx, rowIDs, columnIDs []uint64, options *ImportOptions) error { f.mu.Lock() defer f.mu.Unlock() - p := parallelSlices{cols: columnIDs, rows: rowIDs} - p.fullPrune() - columnIDs = p.cols - rowIDs = p.rows + // if ingest promises that this is "fully sorted", then we have been + // promised that (1) there's no duplicate entries that need to be + // 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 + } // create a mask of columns we care about columns := roaring.NewSliceBitmap(columnIDs...) @@ -2551,6 +2561,10 @@ func (f *fragment) bulkImportMutex(tx Tx, rowIDs, columnIDs []uint64) error { // positions are sorted by columns, but not by absolute // position. we might want them sorted, though. if pos < prev { + if options.fullySorted { + fmt.Printf("HELP! was promised fully sorted input, but previous position was %d, now generated %d\n", + prev, pos) + } unsorted = true } prev = pos @@ -2581,6 +2595,32 @@ func (f *fragment) bulkImportMutex(tx Tx, rowIDs, columnIDs []uint64) error { return errors.Wrap(f.importPositions(tx, toSet, toClear, rowSet), "importing positions") } +// ClearRecords deletes all bits for the given records. It's basically +// the remove-only part of setting a mutex. +func (f *fragment) ClearRecords(tx Tx, recordIDs []uint64) error { + f.mu.Lock() + defer f.mu.Unlock() + + // create a mask of columns we care about + columns := roaring.NewSliceBitmap(recordIDs...) + + // we now need to find existing rows for these bits. + rowSet := make(map[uint64]struct{}) + var toClear []uint64 + callback := func(pos uint64) error { + toClear = append(toClear, pos) + rowID := pos / ShardWidth + rowSet[rowID] = struct{}{} + return nil + } + findExisting := roaring.NewBitmapBitmapFilter(columns, callback) + err := tx.ApplyFilter(f.index(), f.field(), f.view(), f.shard, 0, findExisting) + if err != nil { + return errors.Wrap(err, "finding existing positions") + } + return errors.Wrap(f.importPositions(tx, nil, toClear, rowSet), "clearing records") +} + // importValue bulk imports a set of range-encoded values. func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDepth uint64, clear bool) error { f.mu.Lock() diff --git a/go.mod b/go.mod index 02133467c..e590bdd1b 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/HdrHistogram/hdrhistogram-go v1.1.0 // indirect github.com/beevik/ntp v0.3.0 github.com/benbjohnson/immutable v0.3.0 + github.com/buger/jsonparser v1.1.1 github.com/cespare/xxhash v1.1.0 github.com/davecgh/go-spew v1.1.1 github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f // indirect diff --git a/go.sum b/go.sum index f89f91301..5f4d1be61 100644 --- a/go.sum +++ b/go.sum @@ -42,6 +42,8 @@ github.com/beorn7/perks v1.0.0 h1:HWo1m869IqiPhD389kmkxeTalrjNbbJTC8LXupb+sl0= github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8= github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs= github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84= +github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMUs= +github.com/buger/jsonparser v1.1.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko= github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc= diff --git a/http/client.go b/http/client.go index ce31012d7..000fe5f2a 100644 --- a/http/client.go +++ b/http/client.go @@ -200,6 +200,90 @@ func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error } return rsp.Indexes, nil } + +// 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 nil, errors.Wrap(err, "creating request") + } + + req.Header.Set("Content-Length", strconv.Itoa(len(buf))) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) + + resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true)) + if err != nil { + return nil, errors.Wrap(err, "executing request") + } + defer resp.Body.Close() + 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) + } + // 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 +func (c *InternalClient) IngestOperations(ctx context.Context, uri *pnet.URI, indexName string, buf []byte) error { + if uri == nil { + uri = c.defaultURI + } + u := uri.Path(fmt.Sprintf("/internal/ingest/%s", indexName)) + req, err := http.NewRequest("POST", u, bytes.NewReader(buf)) + if err != nil { + return errors.Wrap(err, "creating request") + } + + req.Header.Set("Content-Length", strconv.Itoa(len(buf))) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) + + resp, err := c.executeRequest(req.WithContext(ctx)) + if err != nil { + return errors.Wrap(err, "executing request") + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return errors.Errorf("unexpected status code: %s", resp.Status) + } + return nil +} + func (c *InternalClient) PostSchema(ctx context.Context, uri *pnet.URI, s *pilosa.Schema, remote bool) error { u := uri.Path(fmt.Sprintf("/schema?remote=%v", remote)) buf, err := json.Marshal(s) diff --git a/http/handler.go b/http/handler.go index 96c7edf98..663ccb0b9 100644 --- a/http/handler.go +++ b/http/handler.go @@ -40,7 +40,7 @@ import ( "github.com/gorilla/handlers" "github.com/gorilla/mux" - "github.com/molecula/featurebase/v2" + pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/encoding/proto" "github.com/molecula/featurebase/v2/logger" "github.com/molecula/featurebase/v2/pql" @@ -430,7 +430,8 @@ func newRouter(handler *Handler) http.Handler { router.HandleFunc("/internal/index/{index}/shards", handler.handleGetIndexAvailableShards).Methods("GET").Name("GetIndexAvailableShards") router.HandleFunc("/internal/nodes", handler.handleGetNodes).Methods("GET").Name("GetNodes") router.HandleFunc("/internal/shards/max", handler.handleGetShardsMax).Methods("GET").Name("GetShardsMax") // TODO: deprecate, but it's being used by the client - + router.HandleFunc("/internal/ingest/{index}", handler.handleIngestData).Methods("POST").Name("PostIngestData") + router.HandleFunc("/internal/schema", handler.handleIngestSchema).Methods("POST").Name("PostIngestSchema") router.HandleFunc("/internal/translate/index/{index}/keys/find", handler.handleFindIndexKeys).Methods("POST").Name("FindIndexKeys") router.HandleFunc("/internal/translate/index/{index}/keys/create", handler.handleCreateIndexKeys).Methods("POST").Name("CreateIndexKeys") router.HandleFunc("/internal/translate/index/{index}/{partition}", handler.handlePostTranslateIndexDB).Methods("POST").Name("PostTranslateIndexDB") @@ -1218,6 +1219,69 @@ func (h *Handler) handleGetPastQueries(w http.ResponseWriter, r *http.Request) { } +func fieldOptionsToFunctionalOpts(opt fieldOptions) []pilosa.FieldOption { + // Convert json options into functional options. + var fos []pilosa.FieldOption + switch opt.Type { + case pilosa.FieldTypeSet: + fos = append(fos, pilosa.OptFieldTypeSet(*opt.CacheType, *opt.CacheSize)) + case pilosa.FieldTypeInt: + if opt.Min == nil { + min := pql.NewDecimal(int64(math.MinInt64), 0) + opt.Min = &min + } + if opt.Max == nil { + max := pql.NewDecimal(int64(math.MaxInt64), 0) + opt.Max = &max + } + fos = append(fos, pilosa.OptFieldTypeInt(opt.Min.ToInt64(0), opt.Max.ToInt64(0))) + case pilosa.FieldTypeDecimal: + scale := int64(0) + if opt.Scale != nil { + scale = *opt.Scale + } + if opt.Min == nil { + min := pql.NewDecimal(int64(math.MinInt64), scale) + opt.Min = &min + } + if opt.Max == nil { + max := pql.NewDecimal(int64(math.MaxInt64), scale) + opt.Max = &max + } + var minmax []pql.Decimal + if opt.Min != nil { + minmax = []pql.Decimal{ + *opt.Min, + } + if opt.Max != nil { + minmax = append(minmax, *opt.Max) + } + } + fos = append(fos, pilosa.OptFieldTypeDecimal(scale, minmax...)) + case pilosa.FieldTypeTimestamp: + if opt.Epoch == nil { + epoch := pilosa.DefaultEpoch + opt.Epoch = &epoch + } + fos = append(fos, pilosa.OptFieldTypeTimestamp(opt.Epoch.UTC(), *opt.TimeUnit)) + case pilosa.FieldTypeTime: + fos = append(fos, pilosa.OptFieldTypeTime(*opt.TimeQuantum, opt.NoStandardView)) + case pilosa.FieldTypeMutex: + fos = append(fos, pilosa.OptFieldTypeMutex(*opt.CacheType, *opt.CacheSize)) + case pilosa.FieldTypeBool: + fos = append(fos, pilosa.OptFieldTypeBool()) + } + if opt.Keys != nil { + if *opt.Keys { + fos = append(fos, pilosa.OptFieldKeys()) + } + } + if opt.ForeignIndex != nil { + fos = append(fos, pilosa.OptFieldForeignIndex(*opt.ForeignIndex)) + } + return fos +} + // handlePostField handles POST /field request. func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) { if !validHeaderAcceptJSON(r.Header) { @@ -1255,66 +1319,7 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) { return } - // Convert json options into functional options. - var fos []pilosa.FieldOption - switch req.Options.Type { - case pilosa.FieldTypeSet: - fos = append(fos, pilosa.OptFieldTypeSet(*req.Options.CacheType, *req.Options.CacheSize)) - case pilosa.FieldTypeInt: - if req.Options.Min == nil { - min := pql.NewDecimal(int64(math.MinInt64), 0) - req.Options.Min = &min - } - if req.Options.Max == nil { - max := pql.NewDecimal(int64(math.MaxInt64), 0) - req.Options.Max = &max - } - fos = append(fos, pilosa.OptFieldTypeInt(req.Options.Min.ToInt64(0), req.Options.Max.ToInt64(0))) - case pilosa.FieldTypeDecimal: - scale := int64(0) - if req.Options.Scale != nil { - scale = *req.Options.Scale - } - if req.Options.Min == nil { - min := pql.NewDecimal(int64(math.MinInt64), scale) - req.Options.Min = &min - } - if req.Options.Max == nil { - max := pql.NewDecimal(int64(math.MaxInt64), scale) - req.Options.Max = &max - } - var minmax []pql.Decimal - if req.Options.Min != nil { - minmax = []pql.Decimal{ - *req.Options.Min, - } - if req.Options.Max != nil { - minmax = append(minmax, *req.Options.Max) - } - } - fos = append(fos, pilosa.OptFieldTypeDecimal(scale, minmax...)) - case pilosa.FieldTypeTimestamp: - if req.Options.Epoch == nil { - epoch := pilosa.DefaultEpoch - req.Options.Epoch = &epoch - } - fos = append(fos, pilosa.OptFieldTypeTimestamp(req.Options.Epoch.UTC(), *req.Options.TimeUnit)) - case pilosa.FieldTypeTime: - fos = append(fos, pilosa.OptFieldTypeTime(*req.Options.TimeQuantum, req.Options.NoStandardView)) - case pilosa.FieldTypeMutex: - fos = append(fos, pilosa.OptFieldTypeMutex(*req.Options.CacheType, *req.Options.CacheSize)) - case pilosa.FieldTypeBool: - fos = append(fos, pilosa.OptFieldTypeBool()) - } - if req.Options.Keys != nil { - if *req.Options.Keys { - fos = append(fos, pilosa.OptFieldKeys()) - } - } - if req.Options.ForeignIndex != nil { - fos = append(fos, pilosa.OptFieldForeignIndex(*req.Options.ForeignIndex)) - } - + fos := fieldOptionsToFunctionalOpts(req.Options) field, err := h.api.CreateField(r.Context(), indexName, fieldName, fos...) if _, ok := err.(pilosa.BadRequestError); ok { http.Error(w, err.Error(), http.StatusBadRequest) @@ -1330,6 +1335,300 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) { resp.write(w, err) } +func (h *Handler) handleIngestData(w http.ResponseWriter, r *http.Request) { + if !validHeaderAcceptJSON(r.Header) { + http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) + return + } + + indexName, ok := mux.Vars(r)["index"] + if !ok { + http.Error(w, "index name is required", http.StatusBadRequest) + return + } + + qcx := h.api.Txf().NewQcx() + err := h.api.IngestOperations(r.Context(), qcx, indexName, r.Body) + + resp := successResponse{h: h, Name: indexName} + resp.write(w, err) +} + +type ingestSpec struct { + IndexName string `json:"index-name"` + IndexAction string `json:"index-action"` + FieldAction string `json:"field-action"` + PrimaryKeyType string `json:"primary-key-type"` + Fields []fieldSpec `json:"fields"` +} + +type fieldSpec struct { + FieldName string `json:"field-name"` + FieldType string `json:"field-type"` + FieldOptions fieldOptionSpec `json:"field-options"` +} + +type fieldOptionSpec struct { + EnforceMutualExclusion bool `json:"enforce-mutual-exclusion"` + CacheType *string `json:"cache-type"` + CacheSize *uint32 `json:"cache-size"` + Scale *int64 `json:"scale"` + Epoch *time.Time `json:"epoch"` + Unit *string `json:"unit"` + TimeQuantum *string `json:"time-quantum"` +} + +func fieldSpecToFieldOption(fSpec fieldSpec) fieldOptions { + opt := fieldOptions{} + // map fSpec type name to pilosa type name + + // a string field could be a string set, mutex or time quantum + if fSpec.FieldOptions.TimeQuantum != nil { + opt.Type = "time" + } else if fSpec.FieldOptions.EnforceMutualExclusion { + opt.Type = "mutex" + } else { + opt.Type = "set" + } + + // for other field types, there's a one-to-one mapping + if fSpec.FieldType != "string" && fSpec.FieldType != "id" { + opt.Type = fSpec.FieldType + } + + if fSpec.FieldType == "string" { + var keys bool = true + opt.Keys = &keys + } + opt.CacheType = fSpec.FieldOptions.CacheType + opt.CacheSize = fSpec.FieldOptions.CacheSize + opt.Scale = fSpec.FieldOptions.Scale + opt.Epoch = fSpec.FieldOptions.Epoch + opt.TimeUnit = fSpec.FieldOptions.Unit + if fSpec.FieldOptions.TimeQuantum != nil { + timeQuantumVal := pilosa.TimeQuantum(*fSpec.FieldOptions.TimeQuantum) + opt.TimeQuantum = &timeQuantumVal + } + + 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) + return + } + + resp := successResponse{h: h} + + dec := json.NewDecoder(r.Body) + dec.DisallowUnknownFields() + schema := ingestSpec{} + // 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 { + 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) + } + } + } + } + } + }() + for dec.More() { + err := dec.Decode(&schema) + if err != nil { + resp.write(w, err) + return + } + 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) + } +} + type postFieldRequest struct { Options fieldOptions `json:"options"` } diff --git a/http/handler_internal_test.go b/http/handler_internal_test.go index b3a3b8550..38117fe94 100644 --- a/http/handler_internal_test.go +++ b/http/handler_internal_test.go @@ -21,7 +21,7 @@ import ( "strings" "testing" - "github.com/molecula/featurebase/v2" + pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/pql" ) diff --git a/http/handler_test.go b/http/handler_test.go index e702d6e09..bfea699d3 100644 --- a/http/handler_test.go +++ b/http/handler_test.go @@ -16,11 +16,14 @@ package http_test import ( "encoding/json" + "fmt" "net" + gohttp "net/http" "testing" - "github.com/molecula/featurebase/v2" + pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/http" + "github.com/molecula/featurebase/v2/server" "github.com/molecula/featurebase/v2/test" ) @@ -80,3 +83,90 @@ func TestMarshalUnmarshalTransactionResponse(t *testing.T) { }) } } + +func TestIngestSchemaHandler(t *testing.T) { + c := test.MustRunCluster(t, 3, + []server.CommandOption{ + server.OptCommandServerOptions(pilosa.OptServerNodeID("node0"), pilosa.OptServerClusterHasher(&test.ModHasher{}))}, + []server.CommandOption{ + server.OptCommandServerOptions(pilosa.OptServerNodeID("node1"), pilosa.OptServerClusterHasher(&test.ModHasher{}))}, + []server.CommandOption{ + server.OptCommandServerOptions(pilosa.OptServerNodeID("node2"), pilosa.OptServerClusterHasher(&test.ModHasher{}))}, + ) + defer c.Close() + + schema := ` +{ + "index-name": "example", + "primary-key-type": "string", + "index-action": "create", + "fields": [ + { + "field-name": "idset", + "field-type": "id", + "field-options": { + "cache-type": "none" + } + }, + { + "field-name": "id", + "field-type": "id", + "field-options": { + "enforce-mutual-exclusion": true + } + }, + { + "field-name": "bool", + "field-type": "bool" + }, + { + "field-name": "stringset", + "field-type": "string", + "field-options": { + "cache-type": "ranked", + "cache-size": 100000 + } + }, + { + "field-name": "string", + "field-type": "string", + "field-options": { + "enforce-mutual-exclusion": true + } + }, + { + "field-name": "int", + "field-type": "int" + }, + { + "field-name": "decimal", + "field-type": "decimal", + "field-options": { + "scale": 2 + } + }, + { + "field-name": "timestamp", + "field-type": "timestamp", + "field-options": { + "epoch": "1996-12-19T16:39:57-08:00", + "unit": "µs" + } + }, + { + "field-name": "quantum", + "field-type": "string", + "field-options": { + "time-quantum": "YMDH" + } + } + ] +} +` + m := c.GetPrimary() + schemaURL := fmt.Sprintf("%s/internal/schema", m.URL()) + resp := test.Do(t, "POST", schemaURL, string(schema)) + if resp.StatusCode != gohttp.StatusOK { + t.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) + } +} diff --git a/ingest/codec.go b/ingest/codec.go new file mode 100644 index 000000000..c966e3f1c --- /dev/null +++ b/ingest/codec.go @@ -0,0 +1,583 @@ +// 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 ingest + +import ( + "fmt" + "io" + "io/ioutil" + "math" + "strconv" + "time" + + "github.com/buger/jsonparser" + "github.com/pkg/errors" +) + +// Featurebase has the following field types as of this writing, we plan to +// support all of them but not all are implemented. +// +// Type Single Signed Timestamp +// set no no no +// time no no yes +// mutex yes no no +// int yes yes no +// decimal yes yes no +// timestamp yes yes no + +type KeyLookupFunc func(...string) (map[string]uint64, error) + +// Codec is a single-use parser which decodes data into columnar vectors. +type Codec interface { + AddSetField(name string, lookup KeyLookupFunc) error + AddTimeQuantumField(name string, lookup KeyLookupFunc) error + AddMutexField(name string, lookup KeyLookupFunc) error + AddBoolField(name string) error + AddIntField(name string, lookup KeyLookupFunc) error + AddDecimalField(name string, scale int64) error + AddTimestampField(name string, scale time.Duration, epoch int64) error + + // Parse data from a reader into the vectors. + // This must only be called once on a codec. + Parse(io.Reader) (*Request, error) +} + +type jsonDecFn func(recID uint64, typ jsonparser.ValueType, data []byte) error + +// applyTranslationFn is a function which applies key lookups to the values +// of an operation, meaning it needs to know whether it's applying them +// to the signed or unsigned values. +type applyTranslationFn func(*FieldOperation, []uint64) error + +type jsonFieldCodec struct { + valueKeys *StringTable + currentOp *FieldOperation + decode jsonDecFn + translate applyTranslationFn + // For timestamp: scale-in-nanoseconds; for instance, if scaleUnit is + // 1,000,000,000, we are storing numbers-of-seconds since the Unix epoch. + // The actual value recorded in BSI will be offset by the field's + // epoch, but we don't need to know that. + // For decimal: Decimal digits of precision. So for instance, with + // scaleUnit 2, "1" is stored as 100 and "1.2" is stored as 120. + scaleUnit int64 + epoch int64 // used only by Timestamp fields + scratch []uint64 // reusable scratch space for sets of values + lookup KeyLookupFunc + fieldType FieldType +} + +// JSONCodec is a Codec which accepts a JSON map of record keys/ids to updated value maps. +type JSONCodec struct { + recKeys *StringTable + fields map[string]*jsonFieldCodec + keyLookup KeyLookupFunc + currentOp *Operation +} + +var _ Codec = &JSONCodec{} + +func NewJSONCodec(lookup KeyLookupFunc) (*JSONCodec, error) { + j := &JSONCodec{fields: map[string]*jsonFieldCodec{}} + if lookup != nil { + j.recKeys = NewStringTable() + j.keyLookup = lookup + } + return j, nil +} + +func (codec *JSONCodec) AddTimeQuantumField(name string, lookup KeyLookupFunc) error { + fieldCodec := &jsonFieldCodec{} + fieldCodec.decode = fieldCodec.DecodeTimeQuantumValue + if lookup != nil { + fieldCodec.translate = (*FieldOperation).TranslateUnsigned + } + return codec.addField(name, FieldTypeTimeQuantum, fieldCodec, lookup) +} + +func (codec *JSONCodec) AddSetField(name string, lookup KeyLookupFunc) error { + fieldCodec := &jsonFieldCodec{} + fieldCodec.decode = fieldCodec.DecodeSetValue + if lookup != nil { + fieldCodec.translate = (*FieldOperation).TranslateUnsigned + } + return codec.addField(name, FieldTypeSet, fieldCodec, lookup) +} + +func (codec *JSONCodec) AddIntField(name string, lookup KeyLookupFunc) error { + fieldCodec := &jsonFieldCodec{} + fieldCodec.decode = fieldCodec.DecodeIntValue + if lookup != nil { + fieldCodec.translate = (*FieldOperation).TranslateSigned + } + return codec.addField(name, FieldTypeInt, fieldCodec, lookup) +} + +func (codec *JSONCodec) AddMutexField(name string, lookup KeyLookupFunc) error { + fieldCodec := &jsonFieldCodec{} + fieldCodec.decode = fieldCodec.DecodeMutexValue + if lookup != nil { + fieldCodec.translate = (*FieldOperation).TranslateUnsigned + } + return codec.addField(name, FieldTypeMutex, fieldCodec, lookup) +} + +func (codec *JSONCodec) AddBoolField(name string) error { + fieldCodec := &jsonFieldCodec{} + fieldCodec.decode = fieldCodec.DecodeBoolValue + return codec.addField(name, FieldTypeBool, fieldCodec, nil) +} + +// TimestampField is used to store seconds since unix epoch. The numeric values +// stored are adjusted based on the given time.Duration; for instance, if the +// time scale is time.Second, then the second after the epoch is stored as 1, +// if it's time.Millisecond, then it's stored as 1000, etcetera. +func (codec *JSONCodec) AddTimestampField(name string, timeScale time.Duration, epoch int64) error { + fieldCodec := &jsonFieldCodec{scaleUnit: int64(timeScale), epoch: epoch} + fieldCodec.decode = fieldCodec.DecodeTimeValue + return codec.addField(name, FieldTypeTimeStamp, fieldCodec, nil) +} + +// AddDecimalField adds a decimal field, which is stored as integer values +// with a scale offset, but parsed as floating point values. For instance, +// with decimalScale=2, `0.01` would store the value 1. +func (codec *JSONCodec) AddDecimalField(name string, decimalScale int64) error { + fieldCodec := &jsonFieldCodec{scaleUnit: int64(math.Pow10(int(decimalScale)))} + fieldCodec.decode = fieldCodec.DecodeDecimalValue + return codec.addField(name, FieldTypeDecimal, fieldCodec, nil) +} + +func (codec *JSONCodec) addField(name string, fieldType FieldType, fieldCodec *jsonFieldCodec, lookup KeyLookupFunc) error { + if _, ok := codec.fields[name]; ok { + return fmt.Errorf("duplicate field %q", name) + } + if lookup != nil { + fieldCodec.valueKeys = NewStringTable() + fieldCodec.lookup = lookup + } + fieldCodec.fieldType = fieldType + codec.fields[name] = fieldCodec + return nil +} + +// decodeSetOrValue decodes a value which might be either an array of values or a +// single value, where values might be string keys or bare numbers, calling cb for +// each value it finds. +func (j *jsonFieldCodec) decodeSetOrValue(dataType jsonparser.ValueType, data []byte, cb func(uint64) error) (err error) { + switch dataType { + case jsonparser.Array: + _, arrayErr := jsonparser.ArrayEach(data, func(value []byte, dataType jsonparser.ValueType, offset int, unused error) { + id, idErr := j.valueKeys.ID(value) + if idErr != nil { + err = idErr + return + } + // stash an error if we got one + valueErr := cb(id) + if valueErr != nil { + err = valueErr + } + }) + if err != nil { + return err + } + if arrayErr != nil { + return arrayErr + } + case jsonparser.String: + id, err := j.valueKeys.ID(data) + if err != nil { + return err + } + return cb(id) + case jsonparser.Number: + if j.valueKeys != nil { + return errors.New("expecting key, got numeric value") + } + id, err := strconv.ParseUint(pretendByteIsString(data), 10, 64) + if err != nil { + return err + } + return cb(id) + default: + return fmt.Errorf("expecting array, got %v", dataType) + } + return err +} + +// DecodeSetValue decodes a set of unsigned values from the provided data into the +// associated currentOp. +func (j *jsonFieldCodec) DecodeSetValue(recID uint64, dataType jsonparser.ValueType, data []byte) (err error) { + return j.decodeSetOrValue(dataType, data, func(id uint64) error { + j.currentOp.AddPair(recID, id) + return nil + }) +} + +// DecodeIntValue decodes a single signed value from the provided data +// into the associated currentOp. +func (j *jsonFieldCodec) DecodeIntValue(recID uint64, dataType jsonparser.ValueType, data []byte) error { + switch dataType { + case jsonparser.String: + value, err := j.valueKeys.IntID(data) + if err != nil { + return err + } + j.currentOp.AddSignedPair(recID, value) + case jsonparser.Number: + if j.valueKeys != nil { + return errors.New("expecting string key, got numeric value") + } + value, err := strconv.ParseInt(pretendByteIsString(data), 10, 64) + if err != nil { + return err + } + j.currentOp.AddSignedPair(recID, value) + default: + if j.valueKeys != nil { + return fmt.Errorf("expecting string key, got %v", dataType) + } else { + return fmt.Errorf("expecting integer value, got %v", dataType) + } + } + return nil +} + +// DecodeMutexValue decodes a single unsigned value from the provided data +// into the associated currentOp. +func (j *jsonFieldCodec) DecodeMutexValue(recID uint64, dataType jsonparser.ValueType, data []byte) error { + switch dataType { + case jsonparser.Number, jsonparser.String: + id, err := j.valueKeys.ID(data) + if err != nil { + return err + } + j.currentOp.AddPair(recID, id) + default: + return fmt.Errorf("expecting integer value, got %v", dataType) + } + return nil +} + +// DecodeBoolValue decodes a single true/false value from the provided data +// into the associated currentOp. +func (j *jsonFieldCodec) DecodeBoolValue(recID uint64, typ jsonparser.ValueType, data []byte) error { + value := uint64(0) + switch typ { + case jsonparser.String: + if string(data) != "true" && string(data) != "false" { + return fmt.Errorf("expecting boolean, got %q", data) + } + // if it's exactly "true" or "false" let's be forgiving + fallthrough + case jsonparser.Boolean: + if data[0] == 't' { + value = 1 + } + case jsonparser.Number: + v, err := jsonparser.GetInt(data) + if err != nil { + return err + } + if v == 1 { + value = 1 + } else if v != 0 { + return errors.New("boolean should be true/false/0/1") + } + default: + return errors.New("boolean should be true/false/0/1") + } + j.currentOp.AddPair(recID, value) + return nil +} + +// DecodeTimeQuantumValue decodes a timestamp, and a set of bits from the +// provided data into the associated currentOp. +func (j *jsonFieldCodec) DecodeTimeQuantumValue(recID uint64, typ jsonparser.ValueType, data []byte) error { + j.scratch = j.scratch[:0] + stamp := time.Unix(0, 0).UTC() + err := jsonparser.ObjectEach(data, func(key []byte, value []byte, dataType jsonparser.ValueType, offset int) (err error) { + switch string(key) { + case "time": + switch dataType { + case jsonparser.String: + stamp, err = time.Parse(time.RFC3339, pretendByteIsString(value)) + case jsonparser.Number: + var unix int64 + unix, err = strconv.ParseInt(pretendByteIsString(value), 10, 64) + stamp = time.Unix(unix, 0) + default: + return fmt.Errorf("expecting time, got %q", value) + } + case "values": + err = j.decodeSetOrValue(dataType, value, func(id uint64) error { + j.scratch = append(j.scratch, id) + return nil + }) + } + return err + }) + if err != nil { + return err + } + if len(j.scratch) > 0 { + defer func() { + // mark these as consumed so if we get called + // again, and don't see a "values" key, we don't reuse them. + j.scratch = j.scratch[:0] + }() + unix := stamp.UnixNano() + for _, value := range j.scratch { + j.currentOp.AddStampedPair(recID, value, unix) + } + } + return nil +} + +// DecodeTimeValue will eventually work but right now it doesn't actually. +func (j *jsonFieldCodec) DecodeTimeValue(recID uint64, dataType jsonparser.ValueType, data []byte) (err error) { + var stamp time.Time + switch dataType { + case jsonparser.String: + stamp, err = time.Parse(time.RFC3339Nano, pretendByteIsString(data)) + if err != nil { + return fmt.Errorf("parsing timestamp: %w", err) + } + j.currentOp.AddSignedPair(recID, (stamp.UnixNano()/j.scaleUnit)-j.epoch) + case jsonparser.Number: + // We could in theory convert this to a time, then convert it + // back, by multiplying by scaleUnit, then dividing. Or... not. + i64, err := strconv.ParseInt(pretendByteIsString(data), 10, 64) + if err != nil { + return fmt.Errorf("parsing numeric timestamp: %w", err) + } + j.currentOp.AddSignedPair(recID, i64) + } + return nil +} + +// DecodeDecimalValue will eventually work but right now it doesn't actually. +func (j *jsonFieldCodec) DecodeDecimalValue(recID uint64, dataType jsonparser.ValueType, data []byte) error { + switch dataType { + case jsonparser.String, jsonparser.Number: + value, err := jsonparser.GetFloat(data) + if err != nil { + return err + } + j.currentOp.AddSignedPair(recID, int64(value*float64(j.scaleUnit))) + default: + return fmt.Errorf("expecting floating-point value, got %v", dataType) + } + + return nil +} + +func (codec *JSONCodec) ParseKeyedRecords(data []byte) (err error) { + seen := make(map[uint64]struct{}) + return jsonparser.ObjectEach(data, func(key []byte, value []byte, dataType jsonparser.ValueType, offset int) error { + id, err := codec.recKeys.ID(key) + if err != nil { + return err + } + if _, ok := seen[id]; ok { + return fmt.Errorf("key %q duplicated in input", key) + } + seen[id] = struct{}{} + if codec.currentOp.OpType == OpWrite { + codec.currentOp.ClearRecordIDs = append(codec.currentOp.ClearRecordIDs, id) + } + return jsonparser.ObjectEach(value, func(key []byte, value []byte, dataType jsonparser.ValueType, offset int) error { + fieldCodec, ok := codec.fields[string(key)] + if !ok { + return errFieldNotFound{string(key)} + } + err := fieldCodec.decode(id, dataType, value) + if err != nil { + return fmt.Errorf("parsing value for field %q: %v", key, err) + } + return nil + }) + }) +} + +func (codec *JSONCodec) ParseOperation(data []byte) (op *Operation, err error) { + op = &Operation{FieldOps: make(map[string]*FieldOperation, len(codec.fields))} + codec.currentOp = op + err = jsonparser.ObjectEach(data, func(key []byte, value []byte, dataType jsonparser.ValueType, offset int) error { + switch string(key) { + case "action": + op.OpType, err = ParseOpType(string(value)) + if err != nil { + return fmt.Errorf("unknown action type %q", value) + } + case "records": + for name, fieldCodec := range codec.fields { + fieldOp := &FieldOperation{} + op.FieldOps[name] = fieldOp + // cache this op so we don't have to do the lookups every time + fieldCodec.currentOp = fieldOp + } + err = codec.ParseKeyedRecords(value) + if err != nil { + return fmt.Errorf("parsing records: %v", err) + } + case "record_ids": + var id uint64 + var idErr error + _, err = jsonparser.ArrayEach(value, func(value []byte, dataType jsonparser.ValueType, offset int, unused error) { + id, idErr = codec.recKeys.ID(value) + if idErr != nil { + return + } + op.ClearRecordIDs = append(op.ClearRecordIDs, id) + + }) + // if we got an error converting an ID, error out with it here. + // we can't stop the ArrayEach early, though? + if idErr != nil { + return idErr + } + if err != nil { + return err + } + case "fields": + _, err = jsonparser.ArrayEach(value, func(value []byte, dataType jsonparser.ValueType, offset int, unused error) { + op.ClearFields = append(op.ClearFields, string(value)) + }) + if err != nil { + return err + } + default: + return fmt.Errorf("unknown operation field %q", key) + } + return nil + }) + if err != nil { + return nil, err + } + if op.OpType == OpNone { + return nil, fmt.Errorf("action not specified") + } + return op, err +} + +// Parse reads a request, but does not sort the results at all or divide +// them into shards. +func (codec *JSONCodec) Parse(r io.Reader) (req *Request, err error) { + data, err := ioutil.ReadAll(r) + if err != nil { + return nil, err + } + 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 + _, err = jsonparser.ArrayEach(data, func(value []byte, dataType jsonparser.ValueType, offset int, err error) { + switch dataType { + case jsonparser.Object: + op, err := codec.ParseOperation(value) + if err != nil { + lastErr = fmt.Errorf("parsing operation: %v", err) + return + } + ops = append(ops, op) + default: + lastErr = fmt.Errorf("expected operation, found %s", dataType) + } + }) + if lastErr != nil { + return nil, lastErr + } + if err != nil { + return nil, err + } + // and now, key translation! + var keyMap []uint64 + if codec.keyLookup != nil { + keyMap, err = MapForStringTable(codec.recKeys, codec.keyLookup) + if err != nil { + return nil, fmt.Errorf("trying to find record key mapping: %w", err) + } + } + req = &Request{FieldTypes: make(map[string]FieldType, len(codec.fields))} + valueMaps := map[string]func(*FieldOperation) error{} + for name, fieldCodec := range codec.fields { + // make closure survive iteration + fieldCodec := fieldCodec + if fieldCodec.lookup != nil { + fieldMap, err := MapForStringTable(fieldCodec.valueKeys, fieldCodec.lookup) + if err != nil { + return nil, fmt.Errorf("trying to find value mapping for %q: %w", name, err) + } + valueMaps[name] = func(fo *FieldOperation) error { + return fieldCodec.translate(fo, fieldMap) + } + } + req.FieldTypes[name] = fieldCodec.fieldType + } + for _, op := range ops { + // For Clear and Write, we need to translate/sort our record ID + // list. + if op.OpType == OpClear || op.OpType == OpWrite || op.OpType == OpDelete { + if keyMap != nil { + if err = translateUnsignedSlice(op.ClearRecordIDs, keyMap); err != nil { + return nil, fmt.Errorf("mapping record keys for clear op: %w", err) + } + } + } + // For clear/delete, that's all we need to do; there's no meaningful fieldops under them. + if op.OpType == OpClear || op.OpType == OpDelete { + 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) + } + } + if fieldTranslate, ok := valueMaps[field]; ok { + if err = fieldTranslate(fieldOp); err != nil { + return nil, fmt.Errorf("mapping values for op on %q: %w", field, err) + } + } + // Sort by column keys, for now. + // Write op will also want to clear every field we saw. + if op.OpType == OpWrite { + op.ClearFields = append(op.ClearFields, field) + } + } + } + req.Ops = ops + return req, nil +} + +type errFieldNotFound struct { + field string +} + +func (err errFieldNotFound) Error() string { + return fmt.Sprintf("field not found: %q", err.field) +} diff --git a/ingest/codec_test.go b/ingest/codec_test.go new file mode 100644 index 000000000..4490d3729 --- /dev/null +++ b/ingest/codec_test.go @@ -0,0 +1,279 @@ +// 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 ingest + +import ( + "fmt" + "testing" + "time" + + "github.com/molecula/featurebase/v2/shardwidth" +) + +func unusableSampleTranslator(keys ...string) (map[string]uint64, error) { + out := make(map[string]uint64, len(keys)) + for _, key := range keys { + out[key] = uint64(len(out)) * 13 + } + return out, nil +} + +func TestSimpleCodec(t *testing.T) { + c, _ := NewJSONCodec(nil) + _ = c.AddSetField("set", nil) + _ = c.AddSetField("setkeys", unusableSampleTranslator) + _ = c.AddMutexField("mutex", nil) + _ = c.AddMutexField("mutexkeys", unusableSampleTranslator) + _ = c.AddTimeQuantumField("tq", nil) + _ = c.AddIntField("int", nil) + _ = 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) + } + _ = c.AddTimestampField("ts", time.Millisecond, epoch.Unix()*1000) + _ = c.AddDecimalField("dec", 2) + _ = c.AddBoolField("bool") + var nextShard = uint64(1< 0 { + subOp.Values = f.Values[prev:endIndex] + } + if len(f.Signed) > 0 { + subOp.Signed = f.Signed[prev:endIndex] + } + target[shard] = subOp + prev = endIndex + } +} + +func ShardIDs(ids []uint64) (out map[uint64][]uint64) { + shards, ends := shardwidth.FindShards(ids) + prev := 0 + out = make(map[uint64][]uint64, len(shards)) + for i, shard := range shards { + endIndex := ends[i] + out[shard] = ids[prev:endIndex] + prev = endIndex + } + return out +} + +// SortToShards() uses a pseudo-radix-sort to divide inputs into +// shards; the individual shards are not sorted. +func (f *FieldOperation) SortToShards() ShardedFieldOperation { + if len(f.RecordIDs) == 0 { + return nil + } + diffMask := uint64(0) + prev := f.RecordIDs[0] + for _, r := range f.RecordIDs[1:] { + diffMask |= r ^ prev + prev = r + } + bitsRemaining := bits.Len64(diffMask) + if bitsRemaining <= shardwidth.Exponent { + return map[uint64]*FieldOperation{f.RecordIDs[0] >> shardwidth.Exponent: f} + } + output := make(ShardedFieldOperation) + sortToShardsInto(f, bitsRemaining-8, output) + return output +} + +// sortToShardsInto puts the shards it finds into the given map, so that +// as we split off buckets, they can be inserted into the same map. +func sortToShardsInto(f *FieldOperation, shift int, into ShardedFieldOperation) { + if shift < shardwidth.Exponent { + shift = shardwidth.Exponent + } + nextShift := shift - 8 + if nextShift < shardwidth.Exponent { + nextShift = shardwidth.Exponent + } + // count things that belong in each of the 256 buckets + var buckets [256]int + var starts [256]int + + // compute the buckets ourselves + for _, r := range f.RecordIDs { + b := (r >> shift) & 0xFF + buckets[b]++ + } + total := 0 + // compute starting points of each bucket, converting the + // bucket counts into ends + for i := range buckets { + starts[i] = total + total += buckets[i] + buckets[i] = total + } + // starts[n] is the index of the first thing that should + // go in that bucket, buckets[n] is the index of the first + // thing that shouldn't + var bucketOp FieldOperation + for bucket, start := range starts { + end := buckets[bucket] + if end <= start { + continue + } + for j := start; j < end; j++ { + want := int((f.RecordIDs[j] >> shift) & 0xFF) + for want != bucket { + // move this to the beginning of the + // bucket it wants to be in, swapping + // the thing there here + dst := starts[want] + f.RecordIDs[j], f.RecordIDs[dst] = f.RecordIDs[dst], f.RecordIDs[j] + if f.Values != nil { + f.Values[j], f.Values[dst] = f.Values[dst], f.Values[j] + } + if f.Signed != nil { + f.Signed[j], f.Signed[dst] = f.Signed[dst], f.Signed[j] + } + starts[want]++ + want = int((f.RecordIDs[j] >> shift) & 0xFF) + } + } + // If shift == shardwidth.Exponent, then this is a completed + // shard and can go into the sharded output. otherwise, we + // can subdivide it. + bucketOp.RecordIDs = f.RecordIDs[start:end] + if f.Values != nil { + bucketOp.Values = f.Values[start:end] + } + if f.Signed != nil { + bucketOp.Signed = f.Signed[start:end] + } + if shift == shardwidth.Exponent { + x := bucketOp + into[f.RecordIDs[start]>>shardwidth.Exponent] = &x + } else { + sortToShardsInto(&bucketOp, nextShift, into) + } + } +} + +const shardMask = ((uint64(1) << shardwidth.Exponent) - 1) + +// SortByValues sorts the operation by values first, then by record +// ID within each value. This is the best ordering for set/mutex fields, +// where we'll want to generate positions in that order. For these +// purposes, a time quantum or bool counts as a kind of a set. +func (f *FieldOperation) SortByValues() { + keys := make([]uint64, len(f.RecordIDs)) + for i, v := range f.RecordIDs { + keys[i] = (f.Values[i] << shardwidth.Exponent) | (v & shardMask) + } + f.SortByKeys(keys) +} + +// SortByRecords sorts the operation by record ID, and not by value at +// all. This makes the most sense for int fields and the like. +func (f *FieldOperation) SortByRecords() { + f.SortByKeys(f.RecordIDs) +} + +// SortByKeys reorganizes the record IDs and values of f according to the +// corresponding members of keys. +func (f *FieldOperation) SortByKeys(keys []uint64) { + if len(f.RecordIDs) < 2 { + return + } + diffMask := uint64(0) + prev := keys[0] + for _, r := range keys[1:] { + diffMask |= r ^ prev + prev = r + } + bitsRemaining := bits.Len64(diffMask) + sortPartialByKeys(f, keys, bitsRemaining-8) +} + +// simpleSort sorts a FieldOperation by external keys, or record IDs. It's a +// horribly naive bubble sort because N is small and a more complex algorithm +// doesn't help as much as you'd hope. This beats using stdlib sort by about +// a factor of two for those small N, for larger N we're using the radix sort +// that calls this. +func simpleSort(f *FieldOperation, keys []uint64) { + if keys != nil { + // sorting by record IDs + if f.Values != nil && f.Signed != nil { + for i := 1; i < len(keys); i++ { + for j := i; j > 0 && keys[j-1] > keys[j]; j-- { + keys[j-1], keys[j] = keys[j], keys[j-1] + f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1] + f.Values[j-1], f.Values[j] = f.Values[j], f.Values[j-1] + f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1] + + } + } + } else if f.Values != nil { + for i := 1; i < len(keys); i++ { + for j := i; j > 0 && keys[j-1] > keys[j]; j-- { + keys[j-1], keys[j] = keys[j], keys[j-1] + f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1] + + f.Values[j-1], f.Values[j] = f.Values[j], f.Values[j-1] + } + } + } else if f.Signed != nil { + for i := 1; i < len(keys); i++ { + for j := i; j > 0 && keys[j-1] > keys[j]; j-- { + keys[j-1], keys[j] = keys[j], keys[j-1] + f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1] + f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1] + } + } + } else { + for i := 1; i < len(keys); i++ { + for j := i; j > 0 && keys[j-1] > keys[j]; j-- { + keys[j-1], keys[j] = keys[j], keys[j-1] + f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1] + } + } + } + } else { + if f.Values != nil && f.Signed != nil { + for i := 1; i < len(f.RecordIDs); i++ { + for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- { + f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1] + f.Values[j-1], f.Values[j] = f.Values[j], f.Values[j-1] + f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1] + } + } + } else if f.Values != nil { + for i := 1; i < len(f.RecordIDs); i++ { + for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- { + f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1] + f.Values[j-1], f.Values[j] = f.Values[j], f.Values[j-1] + } + } + } else if f.Signed != nil { + for i := 1; i < len(f.RecordIDs); i++ { + for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- { + f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1] + f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1] + } + } + } else { + // why do we only have record IDs? I don't know + for i := 1; i < len(f.RecordIDs); i++ { + for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- { + f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1] + } + } + } + } +} + +func sortPartialByKeys(f *FieldOperation, keys []uint64, shift int) { + if shift < 0 { + shift = 0 + } + externalKeys := &f.RecordIDs[0] != &keys[0] + nextShift := shift - 8 + if nextShift < 0 { + nextShift = 0 + } + // count things that belong in each of the 256 buckets + var buckets [256]int + var starts [256]int + // compute the buckets ourselves + for _, r := range keys { + b := (r >> shift) & 0xFF + buckets[b]++ + } + total := 0 + // compute starting points of each bucket, converting the + // bucket counts into ends + for i := range buckets { + starts[i] = total + total += buckets[i] + buckets[i] = total + } + // starts[n] is the index of the first thing that should + // go in that bucket, buckets[n] is the index of the first + // thing that shouldn't + // var newbuckets [256]int + var bucketOp FieldOperation + for bucket, start := range starts { + end := buckets[bucket] + if end <= start { + continue + } + for j := start; j < end; j++ { + want := int((keys[j] >> shift) & 0xFF) + for want != bucket { + // move this to the beginning of the + // bucket it wants to be in, swapping + // the thing there here + dst := starts[want] + keys[j], keys[dst] = keys[dst], keys[j] + // we do this to allow you to just pass in the records as keys + if externalKeys { + f.RecordIDs[j], f.RecordIDs[dst] = f.RecordIDs[dst], f.RecordIDs[j] + } + if f.Values != nil { + f.Values[j], f.Values[dst] = f.Values[dst], f.Values[j] + } + if f.Signed != nil { + f.Signed[j], f.Signed[dst] = f.Signed[dst], f.Signed[j] + } + starts[want]++ + want = int((keys[j] >> shift) & 0xFF) + } + } + // If shift == shardwidth.Exponent, then this is a completed + // shard and can go into the sharded output. otherwise, we + // can subdivide it. + if shift > 0 { + bucketOp.RecordIDs = f.RecordIDs[start:end] + if f.Values != nil { + bucketOp.Values = f.Values[start:end] + } + if f.Signed != nil { + bucketOp.Signed = f.Signed[start:end] + } + // if there's not very many, sort naively instead + if end-start > 32 { + sortPartialByKeys(&bucketOp, keys[start:end], nextShift) + } else { + // naive stdlib sort + if externalKeys { + simpleSort(&bucketOp, keys[start:end]) + } else { + simpleSort(&bucketOp, nil) + } + } + } + } +} + +// AddPair adds a record ID/value pair where the value is unsigned, as +// when used with set/mutex/time quantum fields. +func (f *FieldOperation) AddPair(rec uint64, value uint64) { + f.RecordIDs = append(f.RecordIDs, rec) + f.Values = append(f.Values, value) +} + +// AddSignedPair adds a record ID/value pair where the value is signed, +// as when used with int/decimal/timestamp fields. +func (f *FieldOperation) AddSignedPair(rec uint64, value int64) { + f.RecordIDs = append(f.RecordIDs, rec) + f.Signed = append(f.Signed, value) +} + +// AddStampedPair adds a record/value pair plus a time, which is just +// a Unix time in seconds. (Note, no scaling here; timestamp fields are +// scaled int fields, this is for time quantums.) +func (f *FieldOperation) AddStampedPair(rec uint64, value uint64, stamp int64) { + f.RecordIDs = append(f.RecordIDs, rec) + f.Values = append(f.Values, value) + 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 { + if v >= uint64(len(mapping)) { + oops++ + } else { + target[i] = mapping[v] + } + } + if oops > 0 { + return fmt.Errorf("encountered %d out-of-range keys when applying translation mapping", oops) + } + return nil +} + +// TranslateUnsigned translates keys according to the provided mapping. This +// is used for sets, mutexes, and time quantums. +func (op *FieldOperation) TranslateKeys(mapping []uint64) error { + return translateUnsignedSlice(op.RecordIDs, mapping) +} + +// TranslateUnsigned translates values according to the provided mapping. This +// is used for sets, mutexes, and time quantums. +func (op *FieldOperation) TranslateUnsigned(mapping []uint64) error { + return translateUnsignedSlice(op.Values, mapping) +} + +// TranslateSigned translates signed values according to the provided mapping. +// If we're using this, it's because we're in an integer-type field, which +// admits using keys for fields, so all key values are actually non-negative, +// but the field's type still requires values be expressed as signed ints. +func (op *FieldOperation) TranslateSigned(mapping []uint64) error { + oops := 0 + for i, v := range op.Signed { + if v >= int64(len(mapping)) { + oops++ + } else { + op.Signed[i] = int64(mapping[v]) + } + } + if oops > 0 { + return fmt.Errorf("encountered %d out-of-range signed values when applying translation mapping", oops) + } + return nil +} + +// ShardOperations is a set of Operations associated with a specific shard. +type ShardOperations struct { + Shard uint64 + Ops []*Operation +} + +// Request is a complete ingest request, which may be any combination +// of operations, which may apply to multiple shards. +type Request struct { + FieldTypes map[string]FieldType + Ops []*Operation +} + +// ShardedRequest is an ingest request, split up into individual per-shard +// operations. +type ShardedRequest struct { + FieldTypes map[string]FieldType + Ops map[uint64][]*Operation +} + +// 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 + } + req := make(map[uint64][]*Operation) + shards := make(map[uint64]*Operation) + // we're getting per-field things, which we want to divide per-shard, + // and return to per-shard sets of per-field things, so we're inverting + // the structure. + for _, op := range r.Ops { + // for clear and write ops, we also need to split up the + // ClearRecords values, which may be distinct from the set of + // records for any given field. For Write ops, we'll then end + // up adding in field values for some fields. + if op.OpType == OpClear || op.OpType == OpWrite || op.OpType == OpDelete { + sharded := ShardIDs(op.ClearRecordIDs) + for shard, data := range sharded { + shards[shard] = &Operation{OpType: op.OpType, ClearRecordIDs: data, ClearFields: op.ClearFields, FieldOps: map[string]*FieldOperation{}} + } + } + for field, fieldOp := range op.FieldOps { + sharded := fieldOp.ByShard() + sorter := fieldTypeSorts[r.FieldTypes[field]] + if sorter == nil { + sorter = (*FieldOperation).SortByRecords + } + for shard, data := range sharded { + sorter(data) + shardOp, ok := shards[shard] + if !ok { + if op.OpType == OpWrite { + return nil, fmt.Errorf("write operation has field operation data (%d items) for shard %d, but no clear data", len(data.RecordIDs), shard) + } + shardOp = &Operation{OpType: op.OpType} + shards[shard] = shardOp + shardOp.FieldOps = map[string]*FieldOperation{field: data} + } else { + shardOp.FieldOps[field] = data + } + } + } + for shard, shardOp := range shards { + req[shard] = append(req[shard], shardOp) + } + for k := range shards { + delete(shards, k) + } + + } + return &ShardedRequest{Ops: req}, nil +} + +func (r *Request) Dump(logf func(string, ...interface{})) { + logf("req: %#v", r) + for _, op := range r.Ops { + logf("op: %#v", op) + if len(op.ClearRecordIDs) > 0 { + logf(" clearRecordIDs: %d", op.ClearRecordIDs) + } + if len(op.ClearFields) > 0 { + logf(" clearFields: %s", op.ClearFields) + } + for field, fieldOp := range op.FieldOps { + logf(" field %q: %#v", field, fieldOp) + } + } +} diff --git a/ingest/op_test.go b/ingest/op_test.go new file mode 100644 index 000000000..b0b196e07 --- /dev/null +++ b/ingest/op_test.go @@ -0,0 +1,187 @@ +// 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 ingest_test + +import ( + "math/rand" + "reflect" + "testing" + + "github.com/molecula/featurebase/v2/ingest" + "github.com/molecula/featurebase/v2/shardwidth" +) + +type opShardingTestCase struct { + name string + input ingest.Request + output *ingest.ShardedRequest +} + +var opShardingTestCases = []opShardingTestCase{ + { + name: "sample", + input: ingest.Request{ + Ops: []*ingest.Operation{ + { + OpType: ingest.OpSet, + FieldOps: map[string]*ingest.FieldOperation{ + "shard0": { + RecordIDs: []uint64{0, 1}, + }, + "shard0-1": { + RecordIDs: []uint64{ + 0, + 1 << shardwidth.Exponent, + }, + }, + "shard1": { + RecordIDs: []uint64{ + 1 << shardwidth.Exponent, + 1<>shardwidth.Exponent]++ + } + + sharded := f.SortToShards() + for shard, data := range sharded { + if len(data.RecordIDs) != shards[shard] { + t.Errorf("shard %d: expected %d items, got %d", shard, shards[shard], len(data.RecordIDs)) + } + for _, v := range data.RecordIDs { + if (v >> shardwidth.Exponent) != shard { + t.Errorf("shard %d: got %x, which should be in %d", shard, v, v>>shardwidth.Exponent) + } + } + // expect sorted-ness + data.SortByRecords() + prev := data.RecordIDs[0] + for i, next := range data.RecordIDs[1:] { + if next < prev { + t.Errorf("index %d: prev %d, next %d", i+1, prev, next) + } + prev = next + } + data.SortByValues() + prevV, prevRec := data.Values[0], data.RecordIDs[0] + for i, nextRec := range data.RecordIDs[1:] { + nextV := data.Values[i+1] + if nextV < prevV { + t.Errorf("index %d: prev value %d, next value %d", i+1, prevV, nextV) + } + if nextV == prevV { + if nextRec < prevRec { + t.Errorf("index %d, value %d: prev rec %d, next rec %d", i+1, nextV, prevRec, nextRec) + } + } + prevV = nextV + prevRec = nextRec + } + } + } +} diff --git a/ingest/shard.go b/ingest/shard.go new file mode 100644 index 000000000..7ce7ea6dd --- /dev/null +++ b/ingest/shard.go @@ -0,0 +1,15 @@ +// 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 ingest diff --git a/ingest/sort.go b/ingest/sort.go new file mode 100644 index 000000000..af7eaa6d2 --- /dev/null +++ b/ingest/sort.go @@ -0,0 +1,245 @@ +// 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 ingest + +// "math/bits" + +// HERE THERE BE DRAGONS + +// This is some sorting logic Nia was experimenting with, which we aren't currently +// using, but which beat stdlib sort by a factor-of-several on at least some test +// data, so we aren't deleting it just yet. + +// groupIDPairsByShard destructively groups ID pairs by shard. +// The returned slices reference the original pairs slice. +// func groupIDPairsByShard(pairs []IDPair) map[uint64][]IDPair { +// if len(pairs) == 0 { +// return nil +// } +// +// // Sort pairs by shard (in-place radix sort). +// // This may also change the order of pairs within a shard, but that should not matter. +// for { +// // Find the highest bit which needs to be sorted. +// var diffMask uint64 +// prev := pairs[0].RecordID +// for _, v := range pairs[1:] { +// if v.RecordID < prev { +// diffMask |= v.RecordID ^ prev +// } +// +// prev = v.RecordID +// } +// diffLen := bits.Len64(diffMask) +// if diffLen <= shardwidth.Exponent { +// // The pairs are sorted by shard. +// break +// } +// +// // Select a right bit shift index such that the highest unsorted bit moves to the 128's place. +// shift := uint(diffLen) - 8 +// +// // Create a mask that can be used to group values by sorted bits. +// sortedMask := ^uint64(0) << bits.Len64(diffMask) +// +// for i := 0; i < len(pairs); { +// // Select a group of pairs to sort. +// // While doing so, count the pairs within each bucket. +// j := i +// var buckets [256]struct { +// start, end uint +// } +// for group := pairs[i].RecordID & sortedMask; i < len(pairs) && pairs[i].RecordID&sortedMask == group; i++ { +// buckets[uint8(pairs[i].RecordID>>uint64(shift))].end++ +// } +// group := pairs[j:i] +// +// // Assign indices within the group to the buckets. +// { +// var start uint +// for i := range buckets { +// bucket := &buckets[i] +// bucket.start = start +// bucket.end += start +// start = bucket.end +// } +// } +// +// // Split the group into the buckets. +// for i, b := range buckets { +// // There is no need to update the state of the current bucket - we will never reference it again after this. +// i := uint8(i) +// for j := b.start; j < b.end; j++ { +// // This inner loop may run quite a few times for the first few buckets, but no element will be moved more than twice per byte. +// for uint8(group[j].RecordID>>shift) != i { +// // This pair is in the wrong bucket. +// // Swap it into the correct bucket. +// dstBucket := &buckets[uint8(group[j].RecordID>>shift)] +// k := dstBucket.start +// dstBucket.start++ +// group[j], group[k] = group[k], group[j] +// } +// } +// } +// } +// } +// +// // Split the pairs by shard. +// shards := make(map[uint64][]IDPair) +// for i := 0; i < len(pairs); { +// // Select the shard. +// shard := pairs[i].RecordID >> shardwidth.Exponent +// +// // Find all pairs in the shard. +// j := i +// incr := 1 +// for i+incr < len(pairs) && pairs[i+incr].RecordID>>shardwidth.Exponent == shard { +// i += incr +// incr *= 2 +// } +// for ; incr > 0; incr /= 2 { +// if i+incr < len(pairs) && pairs[i+incr].RecordID>>shardwidth.Exponent == shard { +// i += incr +// } +// } +// // that found us the last thing in this shard, so... +// i++ +// +// // Add the shard, referencing the original slice. +// // This sets the cap so that we dont accidentally overwrite other shards data. +// shards[shard] = pairs[j:i:i] +// } +// +// return shards +// } + +// pairsToZigZag converts a set of record-value pairs to Pilosa's zig-zag format. +// This assumes that all pairs are within the same shard. +// func pairsToZigZag(pairs []IDPair) []uint64 { +// const recordMask = (1 << shardwidth.Exponent) - 1 +// +// dst := make([]uint64, len(pairs)) +// for i, p := range pairs { +// dst[i] = (p.ID << shardwidth.Exponent) | (p.RecordID & recordMask) +// } +// +// return dst +// } + +// radixSort64 sorts the data with radix-sort. +// The "shift" is the highest differing bit position, rounded down to a multiple of 8. +// If there are duplicates, this may change the duplicate count for some values. +// func radixSort64(data []uint64, shift uint) { +// if len(data) < 2 { +// return +// } +// if shift <= 8 { +// // The data falls into a 16-bit span, so the remaining digits can be sorted simultaneously with a bitmask. +// maskSort(data) +// return +// } +// +// // Count the values within each bucket. +// var buckets [256]struct { +// start, end uint +// } +// for _, v := range data { +// buckets[uint8(v>>shift)].end++ +// } +// +// // Assign indices within the group to the buckets. +// { +// var start uint +// for i := range buckets { +// bucket := &buckets[i] +// bucket.start = start +// bucket.end += start +// start = bucket.end +// } +// } +// +// // Split the data into the buckets. +// var start uint +// for i, b := range buckets { +// // Replace misplaced values until the contents of the bucket all have the correct digit. +// i := uint8(i) +// for j := b.start; j < b.end; j++ { +// // This inner loop may run quite a few times for the first few buckets, but it will never run more than once-per-element-per-byte. +// for uint8(data[j]>>shift) != i { +// // This pair is in the wrong bucket. +// // Swap it into the correct bucket. +// dstBucket := &buckets[uint8(data[j]>>shift)] +// k := dstBucket.start +// dstBucket.start++ +// data[j], data[k] = data[k], data[j] +// } +// } +// +// // Sort the contents of the bucket. +// data := data[start:b.end] +// switch { +// case len(data) < 64: +// // Use insertion-sort because the data is too small for a more complex algorithm to be efficient. +// for i := 0; i < len(data); i++ { +// for j := i; j > 0 && data[j-1] > data[j]; j-- { +// data[j-1], data[j] = data[j], data[j-1] +// } +// } +// +// default: +// // Sort the next byte recursively. +// radixSort64(data, shift-8) +// } +// start = b.end +// } +// } + +// maskSort sorts integers using a bitmask. +// The values must all fall within one 16-bit span. +// If there are duplicates, this may change the duplicate count for some values. +// func maskSort(data []uint64) { +// if len(data) < 2 { +// return +// } +// +// base := data[0] &^ ((1 << 16) - 1) +// +// // Dump everything into the mask. +// var mask [(1 << 16) / 64]uint64 +// for _, v := range data { +// mask[uint16(v)/64] |= 1 << (v % 64) +// } +// +// // Scan through the set bits in the mask. +// k := 0 +// for i, w := range mask { +// for w != 0 { +// j := bits.TrailingZeros64(w) +// w &^= 1 << j +// data[k] = 64*uint64(i) + uint64(j) + base +// k++ +// } +// } +// +// if k < len(data) { +// // Copy the ending value to fill up the rest of the space. +// // This happens once for each duplicate value. +// endVal := data[k-1] +// for k < len(data) { +// data[k] = endVal +// k++ +// } +// } +// } diff --git a/ingest/sort_test.go b/ingest/sort_test.go new file mode 100644 index 000000000..231ff61e8 --- /dev/null +++ b/ingest/sort_test.go @@ -0,0 +1,184 @@ +// 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 ingest + +// func BenchmarkSort64(b *testing.B) { +// gen := func(n, width uint64) func() []uint64 { +// var data []uint64 +// var once sync.Once +// return func() []uint64 { +// once.Do(func() { +// data = make([]uint64, n) +// var rng rand.PCGSource +// rng.Seed(9001) +// for i := range data { +// data[i] = rng.Uint64() % width +// } +// }) +// +// return data +// } +// } +// +// algos := []struct { +// name string +// maxn uint64 +// fn func([]uint64) []uint64 +// }{ +// { +// name: "stdlib", +// maxn: 1024 * 1024 * 1024, +// fn: stdSort, +// }, +// { +// name: "heap", +// maxn: 1024 * 1024 * 1024, +// fn: heapSort, +// }, +// { +// name: "radix-insertion-mask", +// maxn: 1024 * 1024 * 1024, +// fn: dedupSort64, +// }, +// } +// +// widths := []struct { +// name string +// width uint64 +// }{ +// {"64", 64}, +// {"1K", 1024}, +// {"64K", 64 * 1024}, +// {"1M", 1024 * 1024}, +// {"16M", 16 * 1024 * 1024}, +// {"128M", 128 * 1024 * 1024}, +// {"1B", 1024 * 1024 * 1024}, +// } +// +// counts := []struct { +// name string +// n uint64 +// }{ +// {"64", 64}, +// {"1K", 1024}, +// {"64K", 64 * 1024}, +// {"1M", 1024 * 1024}, +// {"4M", 4 * 1024 * 1024}, +// {"16M", 16 * 1024 * 1024}, +// {"64M", 64 * 1024 * 1024}, +// {"256M", 256 * 1024 * 1024}, +// } +// +// for _, width := range widths { +// width := width +// b.Run(width.name, func(b *testing.B) { +// for _, count := range counts { +// if count.n > width.width { +// continue +// } +// +// count := count +// b.Run(count.name, func(b *testing.B) { +// datasrc := gen(count.n, width.width) +// for _, alg := range algos { +// if count.n > alg.maxn { +// continue +// } +// +// alg := alg +// b.Run(alg.name, func(b *testing.B) { +// data := datasrc() +// buf := make([]uint64, len(data)) +// b.SetBytes(8 * int64(len(buf))) +// +// b.StopTimer() +// b.ResetTimer() +// +// for i := 0; i < b.N; i++ { +// copy(buf, data) +// b.StartTimer() +// alg.fn(buf) +// b.StopTimer() +// } +// }) +// } +// }) +// } +// }) +// } +// } +// +// func heapSort(data []uint64) []uint64 { +// for i, v := range data { +// for i > 0 && v > data[(i-1)/2] { +// data[i] = data[(i-1)/2] +// i = (i - 1) / 2 +// } +// data[i] = v +// } +// { +// heap := data +// for len(heap) > 1 { +// heap[0], heap[len(heap)-1] = heap[len(heap)-1], heap[0] +// heap = heap[:len(heap)-1] +// i := 0 +// for { +// max := i +// if r := 2*i + 1; r < len(heap) && heap[r] > heap[max] { +// max = r +// } +// if l := 2*i + 2; l < len(heap) && heap[l] > heap[max] { +// max = l +// } +// if max == i { +// break +// } +// +// heap[max], heap[i] = heap[i], heap[max] +// i = max +// } +// } +// } +// +// j := 1 +// prev := data[0] +// for _, v := range data[1:] { +// if v == prev { +// continue +// } +// +// data[j] = v +// prev = v +// } +// +// return data[:j] +// } +// +// func stdSort(data []uint64) []uint64 { +// sort.Slice(data, func(i, j int) bool { return data[i] < data[j] }) +// +// j := 1 +// prev := data[0] +// for _, v := range data[1:] { +// if v == prev { +// continue +// } +// +// data[j] = v +// prev = v +// } +// +// return data[:j] +// } diff --git a/ingest/translate.go b/ingest/translate.go new file mode 100644 index 000000000..7ce7ea6dd --- /dev/null +++ b/ingest/translate.go @@ -0,0 +1,15 @@ +// 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 ingest diff --git a/ingest/update.go b/ingest/update.go new file mode 100644 index 000000000..8101bb683 --- /dev/null +++ b/ingest/update.go @@ -0,0 +1,256 @@ +// 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 ingest + +import ( + "github.com/molecula/featurebase/v2/roaring" +) + +// ShardUpdate is an update request for a shard. +type ShardUpdate struct { + // TODO: include schema version + + // Sets are the updates to set fields. + Sets map[string]SetMatrixUpdate + + // Mutexes are the updates to mutex fields. + Mutexes map[string]MutexMatrixUpdate + + // TimeTensors are the updates to time fields. + TimeTensors map[string]TimeTensorUpdate + + // Ints are the updates to int fields. + Ints map[string]IntUpdate +} + +// Convert the ID vector to a raw update that can be imported. +// All records are assumed to fall within a shard. +// func (vec IDVector) Convert() (MutexMatrixUpdate, error) { +// // Before converting the vector, do a sanity-check for duplicates. +// dedup := make(map[uint64]struct{}, len(vec.Updates)+len(vec.Clears)) +// for _, p := range vec.Updates { +// if _, ok := dedup[p.RecordID]; !ok { +// return MutexMatrixUpdate{}, errors.New("input contains conflicting updates") +// } +// +// dedup[p.RecordID] = struct{}{} +// } +// for _, p := range vec.Clears { +// if _, ok := dedup[p]; !ok { +// return MutexMatrixUpdate{}, errors.New("input contains conflicting updates") +// } +// +// dedup[p] = struct{}{} +// } +// +// // Convert the updates to a bitmap. +// updates, err := idPairsToBitmap(vec.Updates, false) +// if err != nil { +// return MutexMatrixUpdate{}, errors.Wrap(err, "encoding mutex updates") +// } +// +// // Convert the clears to a bitmap. +// clears, err := idListToBitmap(vec.Clears, false) +// if err != nil { +// return MutexMatrixUpdate{}, errors.Wrap(err, "encoding mutex clears") +// } +// +// // Realign clears bitmap to start of shard. +// if min, ok := clears.Min(); ok { +// min &^= (1 << shardwidth.Exponent) - 1 +// clears = clears.OffsetRange(-min, 0, 1< 0 { +// return SetMatrixUpdate{}, errors.New("set adds and removes overlap") +// } +// +// return SetMatrixUpdate{ +// Add: adds, +// Remove: removes, +// Clear: clears, +// }, nil +// } + +// SetMatrixUpdate is an encoded update request for a set-type (set/mutex) view. +type SetMatrixUpdate struct { + // Add is a bitmap to union the matrix against. + Add *roaring.Bitmap + + // Remove is a bitmap to difference out of the matrix. + // No values will be present in both add and remove. + // This bitmap will not include any records in the clear bitmap. + Remove *roaring.Bitmap + + // Clear is a set of records for which all values should be removed. + // This is applied before add operations. + Clear *roaring.Bitmap +} + +// func idPairsToBitmap(pairs []IDPair, allowDup bool) (*roaring.Bitmap, error) { +// return idListToBitmap(pairsToZigZag(pairs), allowDup) +// } +// +// func idListToBitmap(ids []uint64, allowDup bool) (*roaring.Bitmap, error) { +// vals := dedupSort64(ids) +// if len(vals) != len(ids) && !allowDup { +// return nil, errors.New("input contains duplicate values") +// } +// +// // TODO: make the roaring package actually handle this well +// return roaring.NewBitmap(vals...), nil +// } + +// Convert the int vector to a raw update that can be imported. +// All records are assumed to fall within a shard. +// func (vec IntVector) Convert() (IntUpdate, error) { +// // Convert the clears to a bitmap. +// clears, err := idListToBitmap(vec.Clears, false) +// if err != nil { +// return IntUpdate{}, errors.Wrap(err, "encoding int clears") +// } +// +// // Realign clears bitmap to start of shard. +// if min, ok := clears.Min(); ok { +// min &^= (1 << shardwidth.Exponent) - 1 +// clears = clears.OffsetRange(-min, 0, 1< 0 { +// return IntUpdate{}, errors.New("update duplicated with a clear") +// } +// +// return IntUpdate{ +// BSI: updates, +// Clear: clears, +// }, nil +// } + +// IntUpdate is an encoded update request for a BSI (int/timestamp/etc.) view. +type IntUpdate struct { + // BSI is a bitmap of new BSI data to overwrite existing values. + BSI *roaring.Bitmap + + // Clear is a set of records to assign null. + Clear *roaring.Bitmap +} diff --git a/ingest/vec.go b/ingest/vec.go new file mode 100644 index 000000000..082ff8575 --- /dev/null +++ b/ingest/vec.go @@ -0,0 +1,124 @@ +// 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 ingest + +import ( + "fmt" + "reflect" + "strconv" + "unsafe" + + "github.com/pkg/errors" +) + +// StringTable is a mapping of strings to temporary IDs. +// All mapped-to IDs fall in the range [0, len). The zero value, +// a nil map, instead just parses the numbers. +// +// We keep the array of names in creation order because we want reproducibility; +// the first key we see is always key 0. Otherwise, the keys are created in +// arbitrary orders. +type StringTable struct { + names []string + values map[string]uint64 +} + +// NewStringTable just creates a string table with a non-nil map. +func NewStringTable() *StringTable { + return &StringTable{values: map[string]uint64{}} +} + +// unsafe is +func pretendByteIsString(data []byte) (result string) { + dH := (*reflect.SliceHeader)(unsafe.Pointer(&data)) + sH := (*reflect.StringHeader)(unsafe.Pointer(&result)) + sH.Data = dH.Data + sH.Len = dH.Len + return result +} + +// ID returns an ID associated to the string, adding it to the table if it is not already present, +// or parsing an integer if there's no table. +func (tbl *StringTable) ID(in []byte) (uint64, error) { + str := pretendByteIsString(in) + if tbl != nil { + id, ok := tbl.values[str] + if !ok { + id = uint64(len(tbl.values)) + tbl.values[str] = id + tbl.names = append(tbl.names, str) + } + return id, nil + } + return strconv.ParseUint(str, 10, 64) +} + +// SignedID returns an ID associated to the string, adding it to the table if it is not already present, +// or parsing an integer if there's no table. It yields signed values only. +func (tbl *StringTable) IntID(in []byte) (int64, error) { + str := pretendByteIsString(in) + if tbl != nil { + id, ok := tbl.values[str] + if !ok { + id = uint64(len(tbl.values)) + tbl.values[str] = id + tbl.names = append(tbl.names, str) + } + return int64(id), nil + } + return strconv.ParseInt(str, 10, 64) +} + +// MapForStringTable, given a string table mapping strings to consecutive +// integers and a translation function from strings to "real" keys, yields +// a translation/lookup slice. If it cannot translate all the keys, it +// returns an error. +// +// The lookup function corresponds to the FindKeys/CreateKeys methods of +// featurebase translators, by an AMAZING coincidence. +func MapForStringTable(tbl *StringTable, lookup func(...string) (map[string]uint64, error)) ([]uint64, error) { + lookedUp, err := lookup(tbl.names...) + if err != nil { + return nil, err + } + if len(lookedUp) != len(tbl.names) { + return nil, fmt.Errorf("missing keys: expected %d keys, got %d", len(tbl.values), len(lookedUp)) + } + out := make([]uint64, len(tbl.names)) + for i, v := range tbl.names { + out[i] = lookedUp[v] + } + return out, nil +} + +// TimeFormatForUnit returns the time transfer format (between the update encoder and the update applier) with appropriate resolution for a quantum unit. +func TimeFormatForUnit(unit rune) string { + switch unit { + case 'Y': + return "2006" + case 'M': + return "200601" + case 'D': + return "20060102" + case 'H': + return "2006010203" + default: + panic(errors.Errorf("invalid quantum unit: %q", unit)) + } +} + +// TODO: bool + +// TODO: timestamp (just sugar on top of IntVector) 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/roaring/unmarshal_binary.go b/roaring/unmarshal_binary.go index a29bba516..7d8061a99 100644 --- a/roaring/unmarshal_binary.go +++ b/roaring/unmarshal_binary.go @@ -15,6 +15,7 @@ package roaring import ( + "bytes" "errors" "io" "unsafe" @@ -104,6 +105,16 @@ func (b *Bitmap) UnmarshalBinary(data []byte) (err error) { return nil } +func (b *Bitmap) MarshalBinary() ([]byte, error) { + var buf bytes.Buffer + _, err := b.WriteTo(&buf) + if err != nil { + return nil, err + } + + return buf.Bytes(), nil +} + // InspectBinary reads a roaring bitmap, plus a possible ops log, // and reports back on the contents, including distinguishing between // the original ops log and the post-ops-log contents. diff --git a/time.go b/time.go index a17d37726..4253a4d1e 100644 --- a/time.go +++ b/time.go @@ -42,6 +42,13 @@ func (q TimeQuantum) HasDay() bool { return strings.ContainsRune(string(q), 'D') // HasHour returns true if the quantum contains a 'H' unit. func (q TimeQuantum) HasHour() bool { return strings.ContainsRune(string(q), 'H') } +func (q TimeQuantum) Granularity() rune { + var g rune + for _, g = range q { + } + return g +} + // Valid returns true if q is a valid time quantum value. func (q TimeQuantum) Valid() bool { switch q { 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"` }