From 671a0cf5c6d5527eea218eb5f48c6724db37bfba Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 18 May 2021 14:03:53 -0500 Subject: [PATCH 1/7] tweak ImportValue benchmark With timestamps, we probably want to at least check larger BSI fields, so we add that. Also, tweak the interpretation of b.N (making each N count for 10,000 bits) so we can see allocation load at all. But we also reduce the sparse set to be about one bit per 19 bits, because if we do one per 70,000, and are doing field-at-a-time imports, we're getting hundreds of imports to try to match a target of, say, around a million values. We also sort the inputs, because ImportValue is about to start requiring that, since the API does it anyway. Also, extend this to be available on Fields, because field.ImportValue is ALSO doing things which could be inefficient or expensive. --- field_internal_test.go | 36 ++++++++++++++++++++++++-- fragment_internal_test.go | 54 ++++++++++++++++++++++++++++----------- 2 files changed, 73 insertions(+), 17 deletions(-) diff --git a/field_internal_test.go b/field_internal_test.go index 8a9ba7ebe..23f84bc4c 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -203,7 +203,7 @@ type TestField struct { } // NewTestField returns a new instance of TestField d/0. -func NewTestField(t *testing.T, opts FieldOption) *TestField { +func NewTestField(t testing.TB, opts FieldOption) *TestField { path, err := testhook.TempDirInDir(t, *TempDir, "pilosa-field-") if err != nil { t.Fatal(err) @@ -230,7 +230,7 @@ func NewTestField(t *testing.T, opts FieldOption) *TestField { } // OpenField returns a new, opened field at a temporary path. -func OpenField(t *testing.T, opts FieldOption) *TestField { +func OpenField(t testing.TB, opts FieldOption) *TestField { f := NewTestField(t, opts) return f } @@ -509,6 +509,38 @@ func TestBSIGroup_importValue(t *testing.T) { } // loop } +// benchmarkImportValues is a helper function to explore, very roughly, the cost +// of setting values using the special setter used for imports. +func benchmarkFieldImportValues(b *testing.B, qcx *Qcx, bitDepth uint64, f *TestField, cfunc func(uint64) uint64) { + batches := makeBenchmarkImportValueData(b, bitDepth, cfunc) + for _, req := range batches { + err := f.importValue(qcx, req.ColumnIDs, req.Values, &ImportOptions{}) + if err != nil { + b.Fatalf("error importing values: %s", err) + } + } +} + +// Benchmark performance of setValue for BSI ranges. +func BenchmarkField_ImportValue(b *testing.B) { + depths := []uint64{4, 8, 16, 32} + + for _, bitDepth := range depths { + f := OpenField(b, OptFieldTypeInt(0, 1< 0 { + req := ImportValueRequest{ColumnIDs: columns, Values: values} + batches = append(batches, req) + } + b.StartTimer() + return batches +} + // benchmarkImportValues is a helper function to explore, very roughly, the cost // of setting values using the special setter used for imports. func benchmarkImportValues(b *testing.B, tx Tx, bitDepth uint64, f *fragment, cfunc func(uint64) uint64) { - column := uint64(0) - b.StopTimer() - columns := make([]uint64, b.N) - values := make([]int64, b.N) - for i := 0; i < b.N; i++ { - values[i] = int64(i) - columns[i] = column - column = cfunc(column) - } - b.StartTimer() - err := f.importValue(tx, columns, values, bitDepth, false) - if err != nil { - b.Fatalf("error importing values: %s", err) + batches := makeBenchmarkImportValueData(b, bitDepth, cfunc) + for _, req := range batches { + err := f.importValue(tx, req.ColumnIDs, req.Values, bitDepth, false) + if err != nil { + b.Fatalf("error importing values: %s", err) + } } } // Benchmark performance of setValue for BSI ranges. func BenchmarkFragment_ImportValue(b *testing.B) { - depths := []uint64{4, 8, 16} + depths := []uint64{4, 8, 16, 32} for _, bitDepth := range depths { name := fmt.Sprintf("Depth%d", bitDepth) f, idx, tx := mustOpenBSIFragment(b, "i", "f", viewBSIGroupPrefix+"foo", 0) _ = idx b.Run(name+"_Sparse", func(b *testing.B) { - benchmarkImportValues(b, tx, bitDepth, f, func(u uint64) uint64 { return (u + 70000) & (ShardWidth - 1) }) + benchmarkImportValues(b, tx, bitDepth, f, func(u uint64) uint64 { return (u + 19) & (ShardWidth - 1) }) }) f.Clean(b) f, idx, tx = mustOpenBSIFragment(b, "i", "f", viewBSIGroupPrefix+"foo", 0) From bb40d6589f4fb8e6653e854628814fbdfc9a54ee Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 18 May 2021 14:08:32 -0500 Subject: [PATCH 2/7] change addOrRemove to not sort inputs We also implement, but disable for now, a check for sortedness of inputs. This check was useful in development but it's expensive (about 5% of CPU time for large inputs!) and once we've verified that we can make it through tests without triggering it, we're probably fine. --- rbf.go | 79 ++++++++++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 68 insertions(+), 11 deletions(-) diff --git a/rbf.go b/rbf.go index ff2eea25c..f1b4a98c6 100644 --- a/rbf.go +++ b/rbf.go @@ -21,7 +21,6 @@ import ( "io/ioutil" "math" "os" - "sort" "strings" "sync" "sync/atomic" @@ -250,21 +249,79 @@ func (tx *RBFTx) Remove(index, field, view string, shard uint64, a ...uint64) (c return tx.addOrRemove(index, field, view, shard, true, a...) } +// sortedParanoia is a flag to enable a check for unsorted inputs to addOrRemove, +// which is expensive in practice and only really useful occasionally. +const sortedParanoia = false + func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, remove bool, a ...uint64) (changeCount int, err error) { if len(a) == 0 { return 0, nil } - - // have to sort, b/c input is not always sorted. - sort.Slice(a, func(i, j int) bool { return a[i] < a[j] }) + name := rbfName(index, field, view, shard) + // this special case can/should possibly go away, except that it + // turns out to be by far the most common case, and we need to know + // there's at least two items to simplify the check-sorted thing. + if len(a) == 1 { + hi, lo := highbits(a[0]), lowbits(a[0]) + rc, err := tx.tx.Container(name, hi) + if err != nil { + return 0, errors.Wrap(err, "failed to retrieve container") + } + if remove { + if rc.N() == 0 { + return 0, nil + } + rc, chng := rc.Remove(lo) + if !chng { + return 0, nil + } + if rc.N() == 0 { + err = tx.tx.RemoveContainer(name, hi) + } else { + err = tx.tx.PutContainer(name, hi, rc) + } + if err != nil { + return 0, err + } + return 1, nil + } else { + rc, chng := rc.Add(lo) + if !chng { + return 0, nil + } + err = tx.tx.PutContainer(name, hi, rc) + if err != nil { + return 0, err + } + return 1, nil + } + } var lastHi uint64 = math.MaxUint64 // highbits is always less than this starter. var rc *roaring.Container var hi uint64 var lo uint16 + // we can accept sorted either ascending or descending. + sign := a[1] - a[0] + prev := a[0] - sign + sign >>= 63 for i, v := range a { - + // This check is noticably expensive (a few percent in some + // use cases) and as long as it passes occasionally it's probably + // not important to run it all the time, and anyway panic is + // not a good choice outside of testing. + if sortedParanoia { + if (v-prev)>>63 != sign { + explain := fmt.Sprintf("addOrRemove: %d < %d != %d < %d", v, prev, a[1], a[0]) + panic(explain) + } + if v == prev { + explain := fmt.Sprintf("addOrRemove: %d twice", v) + panic(explain) + } + } + prev = v hi, lo = highbits(v), lowbits(v) if hi != lastHi { // either first time through, or changed to a different container. @@ -272,19 +329,19 @@ func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, remove boo if i > 0 { // not first time through, write what we got. if remove && (rc == nil || rc.N() == 0) { - err = tx.RemoveContainer(index, field, view, shard, lastHi) + err = tx.tx.RemoveContainer(name, lastHi) if err != nil { return 0, errors.Wrap(err, "failed to remove container") } } else { - err = tx.PutContainer(index, field, view, shard, lastHi, rc) + err = tx.tx.PutContainer(name, lastHi, rc) if err != nil { return 0, errors.Wrap(err, "failed to put container") } } } // get the next container - rc, err = tx.Container(index, field, view, shard, hi) + rc, err = tx.tx.Container(name, hi) if err != nil { return 0, errors.Wrap(err, "failed to retrieve container") } @@ -305,12 +362,12 @@ func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, remove boo // write the last updates. if remove { if rc == nil || rc.N() == 0 { - err = tx.RemoveContainer(index, field, view, shard, hi) + err = tx.tx.RemoveContainer(name, hi) if err != nil { return 0, errors.Wrap(err, "failed to remove container") } } else { - err = tx.PutContainer(index, field, view, shard, hi, rc) + err = tx.tx.PutContainer(name, hi, rc) if err != nil { return 0, errors.Wrap(err, "failed to put container") } @@ -319,7 +376,7 @@ func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, remove boo if rc == nil || rc.N() == 0 { panic("there should be no way to have an empty bitmap AFTER an Add() operation") } - err = tx.PutContainer(index, field, view, shard, hi, rc) + err = tx.tx.PutContainer(name, hi, rc) if err != nil { return 0, errors.Wrap(err, "failed to put container") } From d2b925d2964892e8a21a0cb0c6da495d7d3402b9 Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 18 May 2021 15:17:29 -0500 Subject: [PATCH 3/7] make importValueSmallWrite faster and also the only path Since we don't always have "snapshots" anymore, the arguable benefit of avoiding the snapshot is reduced, and the primary expense of importPositions has been dramatically reduced as well, so let's just use that all the time, and simplify life. We also want to make it faster. We don't know how many bits there are to set or clear in the input set, but we do know exactly how many bits there are to set AND clear. We can subdivide these into batches by rows, then process each batch by storing sets at the bottom and clears at the top. We can also do batches by columns, reducing the memory overhead of unpacking all the bits at once. (For extra credit, we could alternate set/clear settings, and thus do batches of "the clears from row 0, followed by the clears from row 1" and "the sets from row 1, followed by the sets from row 2", and so on, but this is too fancy.) Every caller of importValue is in fact already providing values with column IDs sorted. As such, we don't need a map for checking the previously-set columns; we just need to check against the previous value. --- fragment.go | 231 ++++++++++++++++------------------------------------ 1 file changed, 70 insertions(+), 161 deletions(-) diff --git a/fragment.go b/fragment.go index d81caa460..c458a4cb3 100644 --- a/fragment.go +++ b/fragment.go @@ -1136,74 +1136,6 @@ func (f *fragment) setValueBase(txOrig Tx, columnID uint64, bitDepth uint64, val return changed, err } -// importSetValue is a more efficient SetValue just for imports. -func (f *fragment) importSetValue(txb *TxBitmap, columnID uint64, bitDepth uint64, value int64, clear bool) (changed int, err error) { // nolint: unparam - // Convert value to an unsigned representation. - uvalue := uint64(value) - if value < 0 { - uvalue = uint64(-value) - } - - for i := uint64(0); i < bitDepth; i++ { - bit, err := f.pos(uint64(bsiOffsetBit+i), columnID) - if err != nil { - return changed, errors.Wrap(err, "getting pos") - } - - if uvalue&(1<= 0 || clear { - if c, err := txb.Remove(p); err != nil { - return changed, errors.Wrap(err, "removing sign from storage") - } else if c { - changed++ - } - } else { - if c, err := txb.Add(p); err != nil { - return changed, errors.Wrap(err, "adding sign to storage") - } else if c { - changed++ - } - } - - return changed, nil -} - // sum returns the sum of a given bsiGroup as well as the number of columns involved. // A bitmap can be passed in to optionally filter the computed columns. func (f *fragment) sum(tx Tx, filter *Row, bitDepth uint64) (sum int64, count uint64, err error) { @@ -2613,57 +2545,6 @@ func (f *fragment) bulkImportMutex(tx Tx, rowIDs, columnIDs []uint64) error { return errors.Wrap(f.importPositions(tx, toSet, toClear, rowSet), "importing positions") } -func (f *fragment) importValueSmallWrite(tx Tx, columnIDs []uint64, values []int64, bitDepth uint64, clear bool) error { - // TODO figure out how to avoid re-allocating these each time. Probably - // possible to store them on the fragment with a capacity based on - // MaxOpN. For now, we know that the total number of bits to be - // set+cleared is len(values)*(bitDepth+1), so we make each slice - // slightly more than half of that to try to avoid reallocation. - toSet := make([]uint64, 0, len(columnIDs)*int(bitDepth+1)*(5/8)) - toClear := make([]uint64, 0, len(columnIDs)*int(bitDepth+1)*(5/8)) - colSet := make(map[uint64]struct{}, len(columnIDs)) - - if err := func() (err error) { - for i := len(columnIDs) - 1; i >= 0; i-- { - columnID, value := columnIDs[i], values[i] - if _, ok := colSet[columnID]; ok { - continue - } - - colSet[columnID] = struct{}{} - toSet, toClear, err = f.positionsForValue(columnID, bitDepth, value, clear, toSet, toClear) - if err != nil { - return errors.Wrap(err, "getting positions for value") - } - } - return nil - }(); err != nil { - errOpenStorage := f.openStorage(true) - if errOpenStorage != nil { - f.Logger.Errorf("failed to import data into fragment: %v", err) - f.Logger.Errorf("recovery with openStorage failed for fragment: %v", errOpenStorage) - f.Logger.Debugf("%s", debug.Stack()) - os.Exit(1) - } - return err - } - rowSet := make(map[uint64]struct{}, bitDepth+1) - for i := uint64(0); i < bitDepth+1; i++ { - rowSet[uint64(i)] = struct{}{} - } - err := f.importPositions(tx, toSet, toClear, rowSet) - if err != nil { - return errors.Wrap(err, "importing positions") - } - - if tx.UseRowCache() { - // Reset the rowCache. - f.rowCache = newSimpleCache() - } - - return nil -} - // 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() @@ -2673,56 +2554,84 @@ func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDep if len(columnIDs) != len(values) { return fmt.Errorf("mismatch of column/value len: %d != %d", len(columnIDs), len(values)) } - - if len(columnIDs)*int(bitDepth+1)+f.opN < f.MaxOpN { - return errors.Wrap(f.importValueSmallWrite(tx, columnIDs, values, bitDepth, clear), "import small write") + positionsByDepth := make([][]uint64, bitDepth+2) + toSetByDepth := make([]int, bitDepth+2) + toClearByDepth := make([]int, bitDepth+2) + batchSize := len(columnIDs) + if batchSize > 65536 { + batchSize = 65536 + } + for i := 0; i < int(bitDepth)+2; i++ { + positionsByDepth[i] = make([]uint64, batchSize) + toClearByDepth[i] = batchSize } - // Process every value. - // If an error occurs then reopen the storage. - if f.storage != nil { - f.storage.OpWriter = nil - } - - var totalChanges int - if err := func() (err error) { - // Build changes into temporary bitmap. - txb := NewTxBitmap(tx, f.index(), f.field(), f.view(), f.shard) - for i := range columnIDs { - columnID, value := columnIDs[i], values[i] - if _, err := f.importSetValue(txb, columnID, bitDepth, value, clear); err != nil { - return errors.Wrapf(err, "importSetValue") + row := 0 + columnID := uint64(0) + value := int64(0) + // arbitrarily set prev to be not equal to the first column ID + // we will encounter. + prev := columnIDs[len(columnIDs)-1] + 1 + for len(columnIDs) > 0 { + downTo := len(columnIDs) - batchSize + if downTo < 0 { + downTo = 0 + } + for i := range positionsByDepth { + toSetByDepth[i] = 0 + toClearByDepth[i] = batchSize + } + for i := len(columnIDs) - 1; i >= downTo; i-- { + columnID, value = columnIDs[i], values[i] + columnID = columnID % ShardWidth + if columnID == prev { + continue + } + prev = columnID + row = 0 + if clear { + toClearByDepth[row]-- + positionsByDepth[row][toClearByDepth[row]] = columnID + } else { + positionsByDepth[row][toSetByDepth[row]] = columnID + toSetByDepth[row]++ + } + row++ + columnID += ShardWidth + if value < 0 { + positionsByDepth[row][toSetByDepth[row]] = columnID + toSetByDepth[row]++ + value *= -1 + } else { + toClearByDepth[row]-- + positionsByDepth[row][toClearByDepth[row]] = columnID + } + row++ + columnID += ShardWidth + for j := 0; j < int(bitDepth); j++ { + if value&1 != 0 { + positionsByDepth[row][toSetByDepth[row]] = columnID + toSetByDepth[row]++ + } else { + toClearByDepth[row]-- + positionsByDepth[row][toClearByDepth[row]] = columnID + } + row++ + columnID += ShardWidth + value >>= 1 } } - // Flush changes in bulk back to the transaction. - return txb.Flush() - }(); err != nil { - errOpenStorage := f.openStorage(true) - if errOpenStorage != nil { - f.Logger.Errorf("failed to import data into fragment: %v", err) - f.Logger.Errorf("recovery with openStorage failed for fragment: %v", errOpenStorage) - f.Logger.Debugf("%s", debug.Stack()) - os.Exit(1) + for i := range positionsByDepth { + err := f.importPositions(tx, positionsByDepth[i][:toSetByDepth[i]], positionsByDepth[i][toClearByDepth[i]:], nil) + if err != nil { + return errors.Wrap(err, "importing positions") + } } - return err - } - // Keep stats accurate. We don't call incrementOpN here because it may - // or may not enqueue a request, which would then be in the queue - // taking up space and otherwise being a possible nuisance, when we're - // about to force a snapshot anyway. - f.opN += totalChanges - f.ops++ - - if tx.UseRowCache() { - // Reset the rowCache. - f.rowCache = newSimpleCache() + columnIDs = columnIDs[:downTo] } - // in theory, this should probably have been queued anyway, but if enough - // of the bits matched existing bits, we'll be under our opN estimate, and - // we want to ensure that the snapshot happens. - return f.holder.SnapshotQueue.Immediate(f) + return nil } // importRoaring imports from the official roaring data format defined at From 7c4b91eef09fd22aadd675a784568b1fc2fa459d Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 18 May 2021 16:03:20 -0500 Subject: [PATCH 4/7] simplify field ImportValue There's only ever one view in importValue, but there's also only ever one shard, because importValue is only called by things called from the API after it has split everything up by shard. --- field.go | 87 ++++++++++++++++++++++++++++---------------------------- index.go | 5 ---- 2 files changed, 43 insertions(+), 49 deletions(-) diff --git a/field.go b/field.go index 2eafcd4fa..3f2c4c534 100644 --- a/field.go +++ b/field.go @@ -1525,6 +1525,9 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []*time. return nil } +// importFloatValue imports floating point values. In current usage, this +// should only ever be called with data for a single shard; the API calls +// around this are splitting it up per shard. func (f *Field) importFloatValue(qcx *Qcx, columnIDs []uint64, values []float64, options *ImportOptions) error { // convert values to int64 values based on scale ivalues := make([]int64, len(values)) @@ -1540,6 +1543,9 @@ func (f *Field) importFloatValue(qcx *Qcx, columnIDs []uint64, values []float64, return f.importValue(qcx, columnIDs, ivalues, options) } +// importFloatValue imports timestamp values. In current usage, this +// should only ever be called with data for a single shard; the API calls +// around this are splitting it up per shard. func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time.Time, options *ImportOptions) error { ivalues := make([]int64, len(values)) bsig := f.bsiGroup(f.name) @@ -1553,8 +1559,17 @@ func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time return f.importValue(qcx, columnIDs, ivalues, options) } -// importValue bulk imports range-encoded value data. +// importValue bulk imports range-encoded value data. This function should +// only be called with data for a single shard; the API calls that wrap +// this handle splitting the data up per-shard. func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, options *ImportOptions) (err0 error) { + // no data to import + if len(columnIDs) == 0 { + return nil + } + if len(values) != len(columnIDs) { + return fmt.Errorf("importValue: mismatch between column IDs and values: %d != %d", len(columnIDs), len(values)) + } viewName := viewBSIGroupPrefix + f.name // Get the bsiGroup so we know bitDepth. bsig := f.bsiGroup(f.name) @@ -1565,13 +1580,10 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, option // We want to determine the required bit depth, in case the field doesn't // have as many bits currently as would be needed to represent these values, // but only if the values are in-range for the field. - var min, max int64 - if len(values) > 0 { - min, max = values[0], values[0] - } + min, max := values[0], values[0] - // Split import data by fragment. - dataByFragment := make(map[importKey]importValueData) + // Check for minimum/maximum in case we need to expand the field's + // stated bit depth. for i := range columnIDs { columnID, value := columnIDs[i], values[i] if value > bsig.Max { @@ -1585,15 +1597,6 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, option if value < min { min = value } - - // Attach value to each bsiGroup view. - for _, name := range []string{viewName} { - key := importKey{View: name, Shard: columnID / ShardWidth} - data := dataByFragment[key] - data.ColumnIDs = append(data.ColumnIDs, columnID) - data.Values = append(data.Values, value) - dataByFragment[key] = data - } } // Determine the highest bit depth required by the min & max. @@ -1612,40 +1615,36 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, option } f.mu.Unlock() - // Import into each fragment. - for key, data := range dataByFragment { - // The view must already exist (i.e. we can't create it) - // because we need to know bitDepth (based on min/max value). - view, err := f.createViewIfNotExists(key.View) - if err != nil { - return errors.Wrap(err, "creating view") - } + // Since all data should be for the same shard, we can just compute + // this from the first value. + shard := columnIDs[0] / ShardWidth - frag, err := view.CreateFragmentIfNotExists(key.Shard) - if err != nil { - return errors.Wrap(err, "creating fragment") - } + view, err := f.createViewIfNotExists(viewName) + if err != nil { + return errors.Wrap(err, "creating view") + } - baseValues := make([]int64, len(data.Values)) - for i, value := range data.Values { - baseValues[i] = value - bsig.Base - } + frag, err := view.CreateFragmentIfNotExists(shard) + if err != nil { + return errors.Wrap(err, "creating fragment") + } - // now we know which shard we discovered. - tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: f.idx, Shard: frag.shard}) - if err != nil { - return err - } - // by deferring, even though we are in loop, we get en-mass commit at once if they all succeed, - // or en-mass rollback if any fail. - defer finisher(&err0) - - if err = frag.importValue(tx, data.ColumnIDs, baseValues, requiredDepth, options.Clear); err != nil { - return err + if bsig.Base != 0 { + for i, v := range values { + values[i] = v - bsig.Base } } - return nil + // now we know which shard we discovered. + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: f.idx, Shard: frag.shard}) + if err != nil { + return err + } + // defer the finisher, so it will check the error returned and + // possibly rollback. + defer finisher(&err0) + + return frag.importValue(tx, columnIDs, values, requiredDepth, options.Clear) } func (f *Field) importRoaring(ctx context.Context, tx Tx, data []byte, shard uint64, viewName string, clear bool) error { diff --git a/index.go b/index.go index 460ef5434..18fa1d49e 100644 --- a/index.go +++ b/index.go @@ -861,11 +861,6 @@ type importData struct { ColumnIDs []uint64 } -type importValueData struct { - ColumnIDs []uint64 - Values []int64 -} - // FormatQualifiedIndexName generates a qualified name for the index to be used with Tx operations. func FormatQualifiedIndexName(index string) string { return fmt.Sprintf("%s\x00", index) From aa4a23b2d9e7ab9112ec35af1e1221178477f253 Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 18 May 2021 16:25:17 -0500 Subject: [PATCH 5/7] generate sorted positions from bulkImportStandard Ensure that positions are sorted, and that we don't generate the same position more than once. --- fragment.go | 40 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 38 insertions(+), 2 deletions(-) diff --git a/fragment.go b/fragment.go index c458a4cb3..592701331 100644 --- a/fragment.go +++ b/fragment.go @@ -2215,6 +2215,34 @@ func (f *fragment) bulkImport(tx Tx, rowIDs, columnIDs []uint64, options *Import return f.bulkImportStandard(tx, rowIDs, columnIDs, options) } +// rowColumnSet is a sortable set of row and column IDs which +// correspond, allowing us to ensure that we produce values in +// a predictable order +type rowColumnSet struct { + r []uint64 + c []uint64 +} + +func (r rowColumnSet) Len() int { + return len(r.r) +} + +func (r rowColumnSet) Swap(i, j int) { + r.r[i], r.r[j] = r.r[j], r.r[i] + r.c[i], r.c[j] = r.c[j], r.c[i] +} + +// Sort by row ID first, column second, to sort by fragment position +func (r rowColumnSet) Less(i, j int) bool { + if r.r[i] < r.r[j] { + return true + } + if r.r[i] > r.r[j] { + return false + } + return r.c[i] < r.c[j] +} + // bulkImportStandard performs a bulk import on a standard fragment. May mutate // its rowIDs and columnIDs arguments. func (f *fragment) bulkImportStandard(tx Tx, rowIDs, columnIDs []uint64, options *ImportOptions) (err error) { @@ -2226,13 +2254,21 @@ func (f *fragment) bulkImportStandard(tx Tx, rowIDs, columnIDs []uint64, options lastRowID := uint64(1 << 63) // 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++ { rowID, columnID := rowIDs[i], columnIDs[i] + if rowID == prevRow && columnID == prevCol { + continue + } + prevRow, prevCol = rowID, columnID pos, err := f.pos(rowID, columnID) if err != nil { return err } - columnIDs[i] = pos + columnIDs[next] = pos + next++ // Add row to rowSet. if rowID != lastRowID { @@ -2240,7 +2276,7 @@ func (f *fragment) bulkImportStandard(tx Tx, rowIDs, columnIDs []uint64, options rowSet[rowID] = struct{}{} } } - positions := columnIDs + positions := columnIDs[:next] f.mu.Lock() defer f.mu.Unlock() if options.Clear { From 7572acb4505f6726fbcb42c000e99463914d83a0 Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 18 May 2021 16:36:25 -0500 Subject: [PATCH 6/7] drop "another shard" test as it's probably not valid We've got a fairly consistent thing of the API splitting data up into shards before sending it to a field, which it has to do because of clustering, so we don't intend to support the case where you have data from another shard in a data set. Also drop the identical but mislabeled test from TestIntField's corresponding case. --- field_internal_test.go | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/field_internal_test.go b/field_internal_test.go index 23f84bc4c..337909494 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -589,13 +589,6 @@ func TestIntField_MinMaxForShard(t *testing.T) { expMax: ValCount{Val: 20, Count: 2}, expMin: ValCount{Val: 10, Count: 3}, }, - { - name: "middlevals", - columnIDs: []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100000000, 100000001}, - values: []int64{10, 20, 10, 10, 20, 11, 12, 11, 13, 11, 44, 1}, - expMax: ValCount{Val: 20, Count: 2}, - expMin: ValCount{Val: 10, Count: 3}, - }, } { t.Run(test.name+strconv.Itoa(i), func(t *testing.T) { if err := f.importValue(qcx, test.columnIDs, test.values, options); err != nil { @@ -756,13 +749,6 @@ func TestDecimalField_MinMaxForShard(t *testing.T) { expMax: ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2}, expMin: ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3}, }, - { - name: "another shard", - columnIDs: []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100000000, 100000001}, - values: []float64{10.1, 20.2, 10.1, 10.1, 20.2, 11, 12, 11, 13, 11, 44.39, 0.23}, - expMax: ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2}, - expMin: ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3}, - }, } { t.Run(test.name+strconv.Itoa(i), func(t *testing.T) { if err := f.importFloatValue(qcx, test.columnIDs, test.values, options); err != nil { From 4bad5defb60922fb4ba82360499a1bd7c42160d3 Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 19 May 2021 13:12:33 -0500 Subject: [PATCH 7/7] sort import values stably without using sort.Stable sort.Stable has horrible runtime -- O(n*logn*logn) -- but if we don't use sort.Stable, our logic for ensuring that we apply the "last" value for a given column is actually completely wrong in the first place. --- api.go | 8 ++++++++ handler.go | 20 ++++++++++++++++++-- 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/api.go b/api.go index 75becc899..37d385791 100644 --- a/api.go +++ b/api.go @@ -1568,7 +1568,15 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu } if !options.Presorted { + // horrible hackery: we implement a secondary key so we can + // get a stable sort without using sort.Stable + req.scratch = make([]int, len(req.ColumnIDs)) + for i := range req.scratch { + req.scratch[i] = i + } sort.Sort(req) + // don't keep that list around since we don't need it anymore + req.scratch = nil } isLocalQcx := false if qcx == nil { diff --git a/handler.go b/handler.go index 6276ac535..ee9a7d8b5 100644 --- a/handler.go +++ b/handler.go @@ -123,6 +123,7 @@ type ImportValueRequest struct { TimestampValues []time.Time StringValues []string Clear bool + scratch []int // scratch space to allow us to get a stable sort in reasonable time } // AtomicRecord applies all its Ivr and Ivr atomically, in a Tx. @@ -138,8 +139,20 @@ type AtomicRecord struct { Ir []*ImportRequest // other field types, e.g. single bit } -func (ivr *ImportValueRequest) Len() int { return len(ivr.ColumnIDs) } -func (ivr *ImportValueRequest) Less(i, j int) bool { return ivr.ColumnIDs[i] < ivr.ColumnIDs[j] } +func (ivr *ImportValueRequest) Len() int { return len(ivr.ColumnIDs) } +func (ivr *ImportValueRequest) Less(i, j int) bool { + if ivr.ColumnIDs[i] < ivr.ColumnIDs[j] { + return true + } + if ivr.ColumnIDs[i] > ivr.ColumnIDs[j] { + return false + } + if len(ivr.scratch) > 0 { + return ivr.scratch[i] < ivr.scratch[j] + } + return false +} + func (ivr *ImportValueRequest) Swap(i, j int) { ivr.ColumnIDs[i], ivr.ColumnIDs[j] = ivr.ColumnIDs[j], ivr.ColumnIDs[i] if len(ivr.Values) > 0 { @@ -151,6 +164,9 @@ func (ivr *ImportValueRequest) Swap(i, j int) { } else if len(ivr.StringValues) > 0 { ivr.StringValues[i], ivr.StringValues[j] = ivr.StringValues[j], ivr.StringValues[i] } + if len(ivr.scratch) > 0 { + ivr.scratch[i], ivr.scratch[j] = ivr.scratch[j], ivr.scratch[i] + } } // Validate ensures that the payload of the request is valid.