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

Prototype ingest API

Browse source 
This is the prototype of the new JSON ingest API. It's not for external use yet, it's still experimental.
This commit is contained in:
seebs 2021-08-20 13:26:36 -05:00 • committed by GitHub
commit 397f90896b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
29 changed files with 4345 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

201
api.go
View file

@ -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: {},
}

176
api_test.go
View file

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

23
field.go
View file

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

56
fragment.go
View file

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

1
go.mod
View file

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

2
go.sum
View file

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

84
http/client.go
View file

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

423
http/handler.go
View file

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

2
http/handler_internal_test.go
View file

@ -21,7 +21,7 @@ import (
"strings"
"testing"
"github.com/molecula/featurebase/v2"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/pql"
)

92
http/handler_test.go
View file

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

583
ingest/codec.go Normal file
View file

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

279
ingest/codec_test.go Normal file
View file

@ -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<<shardwidth.Exponent) + 5
sampleJson := []byte(fmt.Sprintf(`
[
{
"action": "set",
"records": {
"2": {
"set": [ 2 ],
"tq": {
"time": "2006-01-02T15:04:05.999999999Z",
"values": [ 6 ]
},
"dec": 1.02,
"int": 3,
"mutex": 4,
"mutexkeys": "key-a",
"intkeys": "key-a"
},
"%d": {
"set": [ 3 ],
"bool": true
},
"1": {
"set": [ 2 ],
"bool": false,
"setkeys": [
"key-a",
"key-b"
],
"tq": { "values": [ 3, 4 ] }
}
}
},
{
"action": "clear",
"record_ids": [ 5, 6, 7 ],
"fields": [ "tq" ]
},
{
"action": "write",
"records": {
"3": {
"set": 2,
"mutex": 4,
"mutexkeys": "key-a",
"tq": {
"time": "2006-01-02T15:04:05.999999999Z",
"values": 6
},
"ts": "2020-01-01T00:01:00.000000000Z",
"int": 3,
"intkeys": "key-a"
},
"4": {
"set": [ 3 ],
"ts": 1577836860000
},
"5": {
"set": [ 2 ],
"setkeys": [ "key-a", "key-b" ],
"tq": { "values": [ 3, 4 ] }
}
}
},
{
"action": "delete",
"record_ids": [ 9 ]
},
{
"action": "set",
"records": {
"2": {
"mutex": 5,
"mutexkeys": "key-b"
}
}
}
]
`, nextShard))
var expected = map[uint64][]*Operation{
0: {
{
OpType: OpSet,
FieldOps: map[string]*FieldOperation{
"mutex": {
RecordIDs: []uint64{2},
Values: []uint64{4},
},
"set": {
RecordIDs: []uint64{1, 2},
Values: []uint64{2, 2},
},
"tq": {
RecordIDs: []uint64{1, 1, 2},
Values: []uint64{3, 4, 6},
Signed: []int64{0, 0, 1136214245999999999},
},
"mutexkeys": {
RecordIDs: []uint64{2},
Values: []uint64{0},
},
"int": {
RecordIDs: []uint64{2},
Signed: []int64{3},
},
"intkeys": {
RecordIDs: []uint64{2},
Signed: []int64{0},
},
"setkeys": {
RecordIDs: []uint64{1, 1},
Values: []uint64{0, 13},
},
"dec": {
RecordIDs: []uint64{2},
Signed: []int64{102},
},
"bool": {
RecordIDs: []uint64{1},
Values: []uint64{0},
},
},
},
{
OpType: OpClear,
ClearRecordIDs: []uint64{5, 6, 7},
ClearFields: []string{"tq"},
},
{
OpType: OpWrite,
ClearRecordIDs: []uint64{3, 4, 5},
ClearFields: []string{"int", "intkeys", "mutex", "mutexkeys", "set", "setkeys", "tq", "ts"},
FieldOps: map[string]*FieldOperation{
"mutex": {
RecordIDs: []uint64{3},
Values: []uint64{4},
},
"set": {
RecordIDs: []uint64{3, 5, 4},
Values: []uint64{2, 2, 3},
},
"tq": {
RecordIDs: []uint64{5, 5, 3},
Values: []uint64{3, 4, 6},
Signed: []int64{0, 0, 1136214245999999999},
},
"mutexkeys": {
RecordIDs: []uint64{3},
Values: []uint64{0},
},
"int": {
RecordIDs: []uint64{3},
Signed: []int64{3},
},
"intkeys": {
RecordIDs: []uint64{3},
Signed: []int64{0},
},
"setkeys": {
RecordIDs: []uint64{5, 5},
Values: []uint64{0, 13},
},
"ts": {
RecordIDs: []uint64{3, 4},
Signed: []int64{60000, 1577836860000},
},
},
},
{
OpType: OpDelete,
ClearRecordIDs: []uint64{9},
},
{
OpType: OpSet,
FieldOps: map[string]*FieldOperation{
"mutex": {
RecordIDs: []uint64{2},
Values: []uint64{5},
},
"mutexkeys": {
RecordIDs: []uint64{2},
Values: []uint64{13},
},
},
},
},
1: {
{
OpType: OpSet,
FieldOps: map[string]*FieldOperation{
"set": {
RecordIDs: []uint64{nextShard},
Values: []uint64{3},
},
"bool": {
RecordIDs: []uint64{nextShard},
Values: []uint64{1},
},
},
},
},
}
req, err := c.ParseBytes(sampleJson)
if err != nil {
t.Fatalf("parsing sample buffer: %v", err)
}
// req.Dump(t.Logf)
sharded, err := req.ByShard()
if err != nil {
t.Errorf("sharding err: %v", err)
}
for shard, ops := range sharded.Ops {
for i, op := range ops {
op.Sort()
for field, fop := range op.FieldOps {
sorter := fieldTypeSorts[req.FieldTypes[field]]
if sorter == nil {
sorter = (*FieldOperation).SortByRecords
}
sorter(fop)
}
var expectedOp *Operation
if i < len(expected[shard]) {
expectedOp = expected[shard][i]
}
if err = op.Compare(expectedOp); err != nil {
t.Errorf("shard %d, op %d: %v", shard, i, err)
}
}
}
}

30
ingest/doc.go Normal file
View file

@ -0,0 +1,30 @@
// 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 provides tooling for accepting record-oriented data updates
// and converting them to data that can be efficiently merged into stored
// data. Nia's original description:
//
// but the overall pipeline is:
// 1. fetch the schema and use it to configure the codec
// 2. parse the data with the codec into vectors, while stuffing temp record key mappings into a string table
// 3. call *CreateKeys on the cluster for all of the things
// 4. generate an ID remapping table for record keys and apply it to all of the vectors
// 5. remap the string keys
// 6. group each vector by shard
// 7. convert the shard vectors into matrix updates
// 8. combine those matrix updates into a shard update
// 9. send the shard updates out over the internal client
// 10. the nodes apply them to RBF
package ingest

714
ingest/op.go Normal file
View file

@ -0,0 +1,714 @@
// 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"
"math/bits"
"sort"
"github.com/molecula/featurebase/v2/shardwidth"
)
type OpType uint8
const (
OpNone = OpType(iota)
OpSet
OpRemove
OpClear
OpWrite
OpDelete
)
var opNames = []string{
"none",
"set",
"remove",
"clear",
"write",
"delete",
}
type FieldType string
const (
FieldTypeSet = "set"
FieldTypeInt = "int"
FieldTypeTimeQuantum = "time"
FieldTypeTimeStamp = "timestamp"
FieldTypeDecimal = "decimal"
FieldTypeMutex = "mutex"
FieldTypeBool = "bool"
)
var fieldTypeSorts = map[FieldType]func(*FieldOperation){
FieldTypeSet: (*FieldOperation).SortByValues,
FieldTypeInt: (*FieldOperation).SortByRecords,
FieldTypeTimeQuantum: (*FieldOperation).SortByValues,
FieldTypeDecimal: (*FieldOperation).SortByRecords,
FieldTypeMutex: (*FieldOperation).SortByValues,
FieldTypeTimeStamp: (*FieldOperation).SortByRecords,
FieldTypeBool: (*FieldOperation).SortByValues,
}
func (o OpType) String() string {
if int(o) < len(opNames) {
return opNames[o]
}
return fmt.Sprintf("invalid-optype-%d", o)
}
func ParseOpType(s string) (OpType, error) {
for i, v := range opNames[1:] {
if s == v {
return OpType(i + 1), nil
}
}
return 0, fmt.Errorf("unknown operation type %q", s)
}
// Operation represents a single set of changes to make to
// the stored data, which means some combination of clearing
// columns, clearing individual bits, or setting bits or values.
// The same data structure can be used whether this represents the
// whole database operation or a single shard's values.
//
// Operations can specify individual per-field operations, which
// have maps of record IDs to values. They can also have a set of
// record IDs and fields to clear. A Clear operation will have only
// record IDs and fields, a Set or Remove will have only FieldOps,
// and a Write will have both -- populating the record IDs and fields
// from the fieldops.
type Operation struct {
OpType OpType
ClearRecordIDs []uint64
ClearFields []string
FieldOps map[string]*FieldOperation
}
// Compare reports whether two operations seem to be the same.
func (got *Operation) Compare(expected *Operation) error {
if got == nil && expected == nil {
return nil
}
if got == nil {
return fmt.Errorf("expected %q op, got nil", expected.OpType)
}
if expected == nil {
return fmt.Errorf("expected no op, got %q", got.OpType)
}
if got.OpType != expected.OpType {
return fmt.Errorf("operation type mismatch: expected %q, got %q", expected.OpType, got.OpType)
}
if len(got.ClearRecordIDs) != len(expected.ClearRecordIDs) {
return fmt.Errorf("clear record counts differ: expected %d, got %d", len(expected.ClearRecordIDs), len(got.ClearRecordIDs))
}
for i, v1 := range got.ClearRecordIDs {
v2 := expected.ClearRecordIDs[i]
if v1 != v2 {
return fmt.Errorf("clear record id %d differs: expected %d, got %d", i, v2, v1)
}
}
if len(got.ClearFields) != len(expected.ClearFields) {
return fmt.Errorf("clear field counts differ: expected %d (%q), got %d (%q)", len(expected.ClearFields), expected.ClearFields, len(got.ClearFields), got.ClearFields)
}
for i, v1 := range got.ClearFields {
v2 := expected.ClearFields[i]
if v1 != v2 {
return fmt.Errorf("clear field %d differs: expected %q, got %q", i, v2, v1)
}
}
for k, fo1 := range got.FieldOps {
fo2 := expected.FieldOps[k]
if err := fo1.Compare(fo2); err != nil {
return fmt.Errorf("field %q mismatch: %w", k, err)
}
}
for k := range expected.FieldOps {
_, ok := got.FieldOps[k]
if !ok {
return fmt.Errorf("expected op for field %q, but none found", k)
}
}
return nil
}
// FieldOperation is the specific set of changes to make to a given
// field.
//
// For a Clear operation, values can be an empty array. For Set or Remove
// operations on sets, RecordIDs can contain duplicates. Times should
// be empty except for time-quantum fields.
type FieldOperation struct {
RecordIDs []uint64
Values []uint64
// For int/timestamp/decimal, this is the value
// For time-quantum, this is the timestamp
// No field has both signed values and timestamps.
// This is not a place of honor.
Signed []int64
}
// Sort sorts the clear record IDs and field list.
func (o *Operation) Sort() {
// I am aware that this is a crime, but it avoids rewriting
// the code and justifies FieldOperation handling the "only record
// IDs" case.
f := FieldOperation{RecordIDs: o.ClearRecordIDs}
f.SortByRecords()
sort.Strings(o.ClearFields)
}
type ShardedFieldOperation map[uint64]*FieldOperation
// ByShard() divides the FieldOperation's values up into corresponding chunks
// based on the shards of record IDs. Does not further sort IDs within those
// chunks.
func (f *FieldOperation) ByShard() ShardedFieldOperation {
if len(f.RecordIDs) == 0 {
return nil
}
return f.SortToShards()
}
// ShardInto puts the shards it finds into a target map.
func (f *FieldOperation) ShardInto(target ShardedFieldOperation) {
shards, ends := shardwidth.FindShards(f.RecordIDs)
prev := 0
for i, shard := range shards {
endIndex := ends[i]
subOp := &FieldOperation{RecordIDs: f.RecordIDs[prev:endIndex]}
if len(f.Values) > 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)
}
}
}

187
ingest/op_test.go Normal file
View file

@ -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 + 1,
},
},
},
},
{
OpType: ingest.OpRemove,
FieldOps: map[string]*ingest.FieldOperation{
"shard0-2": {
RecordIDs: []uint64{1, 2<<shardwidth.Exponent + 1},
},
},
},
},
},
output: &ingest.ShardedRequest{
Ops: map[uint64][]*ingest.Operation{
0: {
{
OpType: ingest.OpSet,
FieldOps: map[string]*ingest.FieldOperation{
"shard0": {
RecordIDs: []uint64{0, 1},
},
"shard0-1": {
RecordIDs: []uint64{
0,
},
},
},
},
{
OpType: ingest.OpRemove,
FieldOps: map[string]*ingest.FieldOperation{
"shard0-2": {
RecordIDs: []uint64{1},
},
},
},
},
1: {
{
OpType: ingest.OpSet,
FieldOps: map[string]*ingest.FieldOperation{
"shard0-1": {
RecordIDs: []uint64{
1 << shardwidth.Exponent,
},
},
"shard1": {
RecordIDs: []uint64{
1 << shardwidth.Exponent,
1<<shardwidth.Exponent + 1,
},
},
},
},
},
2: {
{
OpType: ingest.OpRemove,
FieldOps: map[string]*ingest.FieldOperation{
"shard0-2": {
RecordIDs: []uint64{2<<shardwidth.Exponent + 1},
},
},
},
},
},
},
},
}
func TestOpSharding(t *testing.T) {
for _, c := range opShardingTestCases {
sharded, err := c.input.ByShard()
if err != nil {
t.Errorf("sharding: unexpected error %v", err)
}
if !reflect.DeepEqual(sharded, c.output) {
t.Fatalf("%s: expected %#v, got %#v", c.name, c.output, sharded)
}
}
}
func TestFancySharding(t *testing.T) {
const shardLimit = 700
const recordCount = 5000
grr := rand.New(rand.NewSource(0))
for i := 0; i < 100; i++ {
f := &ingest.FieldOperation{RecordIDs: make([]uint64, recordCount), Values: make([]uint64, recordCount)}
shards := make([]int, shardLimit)
for j := range f.RecordIDs {
v := uint64(grr.Int63n(shardLimit << shardwidth.Exponent))
f.RecordIDs[j] = v
f.Values[j] = uint64(grr.Int63n(8))
shards[v>>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
}
}
}
}

15
ingest/shard.go Normal file
View file

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

245
ingest/sort.go Normal file
View file

@ -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++
// }
// }
// }

184
ingest/sort_test.go Normal file
View file

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

15
ingest/translate.go Normal file
View file

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

256
ingest/update.go Normal file
View file

@ -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<<shardwidth.Exponent)
// }
//
// return MutexMatrixUpdate{
// Update: updates,
// Clear: clears,
// }, nil
// }
// MutexMatrixUpdate is an encoded update request for a mutex-type view.
type MutexMatrixUpdate struct {
// Update is a bitmap of new mutex values.
// This bitmap will not include any records in the clear bitmap.
Update *roaring.Bitmap
// Clear is a set of records for which all values should be removed.
Clear *roaring.Bitmap
}
// Convert the time vectors to a raw update that can be imported.
// All records are assumed to fall within a shard.
// func (vec TimeIDSetsVector) Convert() (TimeTensorUpdate, error) {
// // Convert the clears to a bitmap.
// clears, err := idListToBitmap(vec.Clears, false)
// if err != nil {
// return TimeTensorUpdate{}, errors.Wrap(err, "encoding time tensor 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<<shardwidth.Exponent)
// }
//
// // Convert the adds to bitmaps.
// quantums := make(map[string]*roaring.Bitmap)
// for t, vec := range vec.Quantums {
// adds, err := idPairsToBitmap(vec, true)
// if err != nil {
// return TimeTensorUpdate{}, errors.Wrap(err, "encoding time adds")
// }
//
// quantums[t] = adds
// }
//
// return TimeTensorUpdate{
// Quantums: quantums,
// Clear: clears,
// }, nil
// }
// TimeTensorUpdate is an encoded update request for a time field within a shard.
type TimeTensorUpdate struct {
// Quantums are the component set matrix adds, grouped by time quantum.
Quantums map[string]*roaring.Bitmap
// Clear is a set of records for which all values should be removed.
// This must not overlap with any values in the remove or clear sections of any quantum.
Clear *roaring.Bitmap
}
// Convert the ID set vector to a raw update that can be imported.
// All records are assumed to fall within a shard.
// func (vec IDSetVector) Convert() (SetMatrixUpdate, error) {
// // Convert the clears to a bitmap.
// clears, err := idListToBitmap(vec.Clears, false)
// if err != nil {
// return SetMatrixUpdate{}, errors.Wrap(err, "encoding set 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<<shardwidth.Exponent)
// }
//
// // Convert the adds to a bitmap.
// adds, err := idPairsToBitmap(vec.Adds, false)
// if err != nil {
// return SetMatrixUpdate{}, errors.Wrap(err, "encoding set adds")
// }
//
// // Verify that the removes do not include the cleared records.
// if clears.Any() {
// for i := 0; i < len(vec.Removes); {
// recID := vec.Removes[i].RecordID
// if clears.Contains(recID) {
// return SetMatrixUpdate{}, errors.New("removed element duplicated with a clear")
// }
//
// i++
// for i < len(vec.Removes) && vec.Removes[i].RecordID == recID {
// i++
// }
// }
// }
//
// // Convert the removes to a bitmap.
// removes, err := idPairsToBitmap(vec.Removes, false)
// if err != nil {
// return SetMatrixUpdate{}, errors.Wrap(err, "encoding set removes")
// }
//
// // Check that no bits are both added and removed.
// if adds.IntersectionCount(removes) > 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<<shardwidth.Exponent)
// }
//
// // Convert updates to a bitmap.
// updates, err := idListToBitmap(intToBSI(vec.Updates), false)
// if err != nil {
// return IntUpdate{}, errors.Wrap(err, "encoding int updates")
// }
//
// // Check that no updated records are also being cleared.
// if updates.IntersectionCount(clears) > 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
}

124
ingest/vec.go Normal file
View file

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

468
ingest_test.go Normal file
View file

@ -0,0 +1,468 @@
// Copyright 2021 Molecula Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa_test
import (
"bytes"
"context"
"encoding/json"
"io/ioutil"
"os"
"path/filepath"
"strconv"
"strings"
"testing"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/http"
"github.com/molecula/featurebase/v2/ingest"
"github.com/molecula/featurebase/v2/server"
"github.com/molecula/featurebase/v2/test"
"github.com/pkg/errors"
)
// For ingest API testing, we want to do tests which have a known
// schema, no existing data before we start, and perform ingests and
// then do queries.
//
// A good starting point for this would be a fairly simple file
// divided into sections which are just the JSON text of the data
// we want to be working with, or the PQL queries we want to run,
// or their expected results.
//
// So, roughly like this:
//
// schema:
// {
// "index-name": "example",
// "primary-key-type": "string",
// "fields": [
// {
// "field-name": "set",
// "field-type": "id",
// "field-options": { "cache-type": "none" }
// }
// ]
// }
// ingest:
// [
// {
// "action": "set",
// "records": {
// "1": {
// "set": [ 2 ],
// }
// }
// }
// ]
// queries:
// Row(set=2):
// [1]
//
// Additionally, the names "schema-error" and "ingest-error" are taken to
// represent a schema, or data set, which is expected to produce an error.
// For instance:
//
// ingest-error:
// [ { "action": puppy }
//
// In this case, it would be considered a test failure if an ingest request
// did NOT fail.
// Lines starting with #
// ingestSchemaPartial represents the only part of a schema we need to
// know about in order to undo its creation of a schema for use in a
// test case.
type ingestSchemaPartial struct {
IndexName string `json:"index-name"`
}
type ingestActionKind int
const (
ingestActionNone = ingestActionKind(iota)
ingestActionSchema
ingestActionIngest
ingestActionSchemaError
ingestActionIngestError
ingestActionQueries
)
var ingestActionKinds = map[string]ingestActionKind{
"schema": ingestActionSchema,
"ingest": ingestActionIngest,
"schema-error": ingestActionSchemaError,
"ingest-error": ingestActionIngestError,
"queries": ingestActionQueries,
}
var ingestActionKindNames = map[ingestActionKind]string{}
type testCaseAction struct {
kind ingestActionKind
comment []byte
lineStart int
lineEnd int
data []byte
}
type liner struct {
data []byte
at int
start int
remaining []byte
line []byte
}
func newLiner(data []byte) *liner {
return &liner{data: data, at: 0, remaining: data}
}
func (l *liner) next() bool {
if len(l.remaining) == 0 {
return false
}
foundNL := true
nextNL := bytes.IndexByte(l.remaining, '\n')
if nextNL == -1 {
foundNL = false
nextNL = len(l.remaining)
}
l.line = l.remaining[:nextNL]
// move past the newline we found, if we found one
if foundNL {
nextNL++
}
l.start, l.at = l.at, l.at+nextNL
l.remaining = l.data[l.at:]
return true
}
func (l *liner) text() (line []byte, start int, end int) {
return l.line, l.start, l.at
}
// parseExpectedResults handles something that looks like
// [1, 2, 3] or ["a", "b", "c"]. It does not handle things like
// quotes within strings, etcetera.
func parseExpectedResults(data []byte) (ints []uint64, keys []string, err error) {
if len(data) < 2 || data[0] != '[' || data[len(data)-1] != ']' {
return nil, nil, errors.New("expecting [] results")
}
words := bytes.Split(data[1:len(data)-1], []byte{','})
for _, word := range words {
word = bytes.TrimSpace(word)
if len(word) == 0 {
return nil, nil, errors.New("found empty word expecting result")
}
if word[0] == '"' {
keys = append(keys, string(word[1:len(word)-1]))
continue
}
v, err := strconv.ParseInt(string(word), 10, 64)
if err != nil {
return nil, nil, err
}
ints = append(ints, uint64(v))
}
if len(ints) > 0 && len(keys) > 0 {
return nil, nil, errors.New("mixed integers and strings are invalid")
}
return ints, keys, err
}
func testQueries(t *testing.T, ctx context.Context, cmd *test.Command, index string, action testCaseAction) {
qcx := cmd.API.Txf().NewQcx()
defer func() {
if err := qcx.Finish(); err != nil {
t.Fatalf("finishing qcx: %v", err)
}
}()
l := newLiner(action.data)
for l.next() {
query, _, _ := l.text()
if !l.next() {
t.Fatalf("processing query list: no expected after %q", query)
}
expected, _, _ := l.text()
ints, keys, err := parseExpectedResults(expected)
if err != nil {
t.Fatalf("processing query list: invalid expected results %q", expected)
}
t.Logf("expecting %q -> %s", query, expected)
res, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: string(query)})
if err != nil {
t.Errorf("query: %v", err)
}
if len(res.Results) != 1 {
t.Fatalf("expected one result per query, got %d results", len(res.Results))
}
var row *pilosa.Row
var ok bool
if row, ok = res.Results[0].(*pilosa.Row); !ok {
t.Fatalf("expected results to be a row")
}
if ints != nil {
cols := row.Columns()
if len(cols) != len(ints) {
t.Fatalf("wrong number of values, expected %d, got %d", len(ints), len(cols))
}
for i := range cols {
if ints[i] != cols[i] {
t.Fatalf("result %d: expected %d, got %d", i, ints[i], cols[i])
}
}
}
// key return value is unpredictable, so...
if keys != nil {
seen := make(map[string]struct{})
for _, k := range keys {
seen[k] = struct{}{}
}
for _, k := range row.Keys {
if _, ok := seen[k]; !ok {
t.Fatalf("unexpected result key %q", k)
}
delete(seen, k)
}
for k := range seen {
t.Fatalf("expected result to contain %q, but did not get it", k)
}
}
}
}
// testOneIngestTestcase runs a set of actions, then cleans up after itself
func testOneIngestTestcase(t *testing.T, ctx context.Context, cmd *test.Command, tcpath string) {
data, err := ioutil.ReadFile(tcpath)
if err != nil {
t.Fatalf("reading %q: %v", tcpath, err)
}
var actions []testCaseAction
var action testCaseAction
l := newLiner(data)
var line []byte
var start, lineStart, lineEnd int
lineCount := 0
for l.next() {
line, lineStart, lineEnd = l.text()
lineCount++
if colon := bytes.IndexByte(line, ':'); colon != -1 {
if kind, ok := ingestActionKinds[string(line[:colon])]; ok {
if action.kind != ingestActionNone {
action.data = data[start:lineStart]
actions = append(actions, action)
action.lineEnd = lineCount - 1
} else {
if lineStart != 0 {
t.Logf("warning: %d bytes with no action type before first action", lineStart)
}
}
start = lineEnd
action.data = nil
action.kind = kind
if line[len(line)-1] == ':' {
action.comment = line[:len(line)-1]
} else {
action.comment = line
}
action.lineStart = lineCount
}
}
}
if action.kind != ingestActionNone {
action.lineEnd = lineCount - 1
action.data = data[start:]
actions = append(actions, action)
}
var mostRecentIndex string
seenIndexes := map[string]struct{}{}
cli := cmd.Client()
created := map[string][]string{}
defer func() {
t.Logf("deleting created indexes/fields:")
for k, v := range created {
if len(v) == 0 {
t.Logf(" index: %q", k)
if err := cmd.API.DeleteIndex(ctx, k); err != nil {
t.Errorf("deleting index %q: %v", k, err)
}
} else {
t.Logf(" fields in %q: %q", k, v)
for _, field := range v {
if err := cmd.API.DeleteField(ctx, k, field); err != nil {
t.Errorf("deleting field %q from %q: %v", field, k, err)
}
}
}
}
}()
noticeCreation := func(newlyCreated map[string][]string) {
for k, v := range newlyCreated {
if len(v) == 0 {
if existing, ok := created[k]; ok {
if len(existing) > 0 {
t.Fatalf("creation reports index %q newly created, but we created fields %q in it previously",
k, existing)
}
}
// create an empty list, indicating that the whole index is
// believed nil
created[k] = nil
continue
}
if existing, ok := created[k]; ok {
if len(existing) == 0 {
// we'll delete this index anyway, don't need to delete fields in it
continue
}
created[k] = append(existing, v...)
continue
}
created[k] = v
}
}
for _, action := range actions {
t.Logf("%s, lines %d-%d", action.comment, action.lineStart, action.lineEnd)
switch action.kind {
case ingestActionSchema:
var scratch ingestSchemaPartial
err = json.Unmarshal(action.data, &scratch)
if err != nil {
t.Fatalf("couldn't parse schema data: %v", err)
}
if scratch.IndexName == "" {
t.Fatalf("test case must provide an index name")
}
// stash the string from the schema, because we
// might need it later
mostRecentIndex = scratch.IndexName
seenIndexes[mostRecentIndex] = struct{}{}
var newlyCreated map[string][]string
newlyCreated, err = cli.IngestSchema(ctx, nil, action.data)
if err != nil {
t.Fatalf("executing schema: %v", err)
}
noticeCreation(newlyCreated)
case ingestActionSchemaError:
var scratch ingestSchemaPartial
err = json.Unmarshal(action.data, &scratch)
if err != nil {
t.Logf("got expected error from schema: %v", err)
break
}
var newlyCreated map[string][]string
newlyCreated, err = cli.IngestSchema(ctx, nil, action.data)
if err != nil {
t.Logf("got expected error from schema: %v", err)
break
}
noticeCreation(newlyCreated)
t.Fatalf("expected error from schema, didn't get it")
case ingestActionIngest:
func() {
qcx := cmd.API.Txf().NewQcx()
var err error
defer func() {
if err == nil {
qcx.Abort()
return
}
if err := qcx.Finish(); err != nil {
t.Fatalf("finishing qcx: %v", err)
}
}()
err = cmd.API.IngestOperations(ctx, qcx, mostRecentIndex, bytes.NewBuffer(action.data))
if err != nil {
t.Fatalf("importing data: %v", err)
}
}()
case ingestActionIngestError:
func() {
qcx := cmd.API.Txf().NewQcx()
var err error
defer func() {
if err == nil {
qcx.Abort()
return
}
if err := qcx.Finish(); err != nil {
t.Fatalf("finishing qcx: %v", err)
}
}()
err = cmd.API.IngestOperations(ctx, qcx, mostRecentIndex, bytes.NewBuffer(action.data))
if err != nil {
t.Logf("got expected error from ingest: %v", err)
return
}
t.Fatalf("expected error from ingest, didn't get it")
}()
case ingestActionQueries:
testQueries(t, ctx, cmd, mostRecentIndex, action)
}
}
}
// TestIngestTestcases reads sample test cases from a test data directory
// and evaluates them.
func TestIngestTestcases(t *testing.T) {
_ = &ingest.Operation{}
var testcases []string
if len(ingestActionKindNames) == 0 {
for k, v := range ingestActionKinds {
ingestActionKindNames[v] = k
}
}
err := filepath.Walk("ingest_testdata", func(path string, info os.FileInfo, err error) error {
if err != nil {
return err
}
if info.IsDir() {
return nil
}
if strings.HasSuffix(path, ".tc") {
testcases = append(testcases, path)
}
return nil
})
if err != nil {
t.Fatalf("looking for test cases: %v", err)
}
if len(testcases) == 0 {
t.Fatalf("no ingest test cases found")
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
c := test.MustRunCluster(t, 1,
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()
coord := c.GetPrimary()
for _, tc := range testcases {
t.Run(strings.TrimSuffix(tc, ".tc"), func(t *testing.T) {
testOneIngestTestcase(t, ctx, coord, tc)
})
}
}

17
ingest_testdata/bool.tc Normal file
View file

@ -0,0 +1,17 @@
schema:
{
"index-name": "example",
"index-action": "create",
"primary-key-type": "uint",
"fields": [
{
"field-name": "tf",
"field-type": "bool"
}
]
}
ingest:
[{"action": "write", "records": {"2": { "tf": true }}}]
queries:
Row(tf=true)
[2]

101
ingest_testdata/expect_errors.tc Normal file
View file

@ -0,0 +1,101 @@
schema-error: fail to create field
{
"index-name": "examplekeys",
"index-action": "create",
"primary-key-type": "string",
"fields": [
{
"field-name": "cookie",
"field-type": "id",
"field-options": { "cache-type": "none" }
},
{
"field-name": "set",
"field-type": "id",
"field-options": { "cache-type": "nun" }
}
]
}
schema: confirm index was not created because field failed
{
"index-name": "examplekeys",
"index-action": "create",
"primary-key-type": "string",
"fields": [
{
"field-name": "set",
"field-type": "id",
"field-options": { "cache-type": "none" }
}
]
}
schema-error: can't recreate index
{
"index-name": "examplekeys",
"index-action": "create",
"primary-key-type": "string",
"fields": [
{
"field-name": "newfield",
"field-type": "id",
"field-options": { "cache-type": "none" }
}
]
}
schema: can add field to existing index
{
"index-name": "examplekeys",
"index-action": "ensure",
"primary-key-type": "string",
"fields": [
{
"field-name": "newfield",
"field-type": "id",
"field-options": { "cache-type": "none" }
}
]
}
schema-error: second field failing in existing index
{
"index-name": "examplekeys",
"index-action": "ensure",
"primary-key-type": "string",
"fields": [
{
"field-name": "addokay",
"field-type": "id",
"field-options": { "cache-type": "none" }
},
{
"field-name": "addfail",
"field-type": "id",
"field-options": { "cache-type": "nun" }
}
]
}
schema: verify that fields we think exist do exist
{
"index-name": "examplekeys",
"index-action": "require",
"primary-key-type": "string",
"fields": [
{
"field-name": "newfield",
"field-type": "id",
"field-options": { "cache-type": "none" }
}
]
}
schema-error: successful field deleted anyway because second field failed
{
"index-name": "examplekeys",
"index-action": "require",
"primary-key-type": "string",
"fields": [
{
"field-name": "addokay",
"field-type": "id",
"field-options": { "cache-type": "none" }
}
]
}

32
ingest_testdata/keyed.tc Normal file
View file

@ -0,0 +1,32 @@
schema:
{
"index-name": "examplekeys",
"index-action": "create",
"primary-key-type": "string",
"fields": [
{
"field-name": "set",
"field-type": "id",
"field-options": { "cache-type": "none" }
}
]
}
ingest:
[
{
"action": "set",
"records": {
"a": {
"set": [ 2 ],
},
"b": {
"set": 3,
}
}
}
]
queries:
Row(set=2)
["a"]
Union(Row(set=3),Row(set=2))
["a","b"]

89
ingest_testdata/sample.tc Normal file
View file

@ -0,0 +1,89 @@
schema:
{
"index-name": "example",
"index-action": "create",
"primary-key-type": "uint",
"fields": [
{
"field-name": "set",
"field-type": "id",
"field-options": { "cache-type": "none" }
}
]
}
ingest:
[
{
"action": "set",
"records": {
"1": {
"set": [ 2 ],
},
"2": {
"set": 3,
}
}
}
]
queries:
Row(set=2)
[1]
Union(Row(set=3),Row(set=2))
[1,2]
schema-error:
{
"index-name": "example",
"primary-key-type": "uint",
"index-action": "require",
"fields": [
{
"field-name": "setkey",
"field-type": "string",
"field-options": { "cache-type": "none" }
}
]
}
schema:
{
"index-name": "example",
"primary-key-type": "uint",
"index-action": "ensure",
"fields": [
{
"field-name": "setkey",
"field-type": "string",
"field-options": { "cache-type": "none" }
}
]
}
ingest:
[
{
"action": "set",
"records": {
"1": {
"setkey": [ "a" ],
},
"2": {
"setkey": "b",
}
}
}
]
queries:
Row(setkey="a")
[1]
ingest-error:
[
{
"action": "setkeys",
"records": {
"a": {
"setkey": [ "a" ],
},
"b": {
"setkey": "b",
}
}
}
]

11
roaring/unmarshal_binary.go
View file

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

7
time.go
View file

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

2
tracker.go
View file

@ -34,7 +34,7 @@ type PastQueryStatus struct {
Node string `json:"nodeID"`
Index string `json:"index"`
Start time.Time `json:"start"`
Runtime time.Duration `json:"runtime"` // deprecated
Runtime time.Duration `json:"runtime"` // deprecated
RuntimeNs time.Duration `json:"runtimeNanoseconds"`
}