scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

make import correctly reflect that it needs a single shard always

Browse source 
In fact, we have a number of things assuming that values passed to Import
always fit within a single known shard, so, drop all the extra complexity
around this, drop the computation of fancy view/shard keys, and so on.

There's a lot of room left to improve this probably but it's at least
better, I think.

Unfortunately, there's a handful of things, basically all of which are
test cases, which were relying on this, so, we also add functionality
for splitting import requests by shards. But this allows us to stop
duplicating each shard's inputs one at a time... which turns out to
mean that we now care that the import operation can write back to the
import request. This only affects test cases, so we adopt a crufty
hack involving cloning import requests in those rare cases, and also
when reusing the same column IDs to write to the existence field that
we'd be using later to write to another field.

Note that even if we weren't overwriting the column IDs with positions,
we'd be sorting the column/row ID lists by row-then-column, which means
we'd still be corrupting the column ID lists. This may want to change
at some point.

We also reuse a single Tx for all the views, because DB-per-shard
means that should work fine, and reduces the cost of doing these
updates, probably.
This commit is contained in:
Seebs 2021-07-26 13:29:55 -05:00 • committed by Seebs
parent 6af987a5ba
commit f019cc7409
10 changed files with 236 additions and 71 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
api.go
View file

@ -1619,7 +1619,7 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest,
// Note: req.Shard may not be the only shard imported into here,
// so don't expect it to be invariant.
if !options.Clear {
if err := importExistenceColumns(qcx, idx, req.ColumnIDs); err != nil {
if err := importExistenceColumns(qcx, idx, req.ColumnIDs, req.Shard); err != nil {
api.server.logger.Errorf("import existence error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
return err
}
@ -1629,7 +1629,7 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest,
}
// Import into fragment.
err = field.Import(qcx, req.RowIDs, req.ColumnIDs, timestamps, opts...)
err = field.Import(qcx, req.RowIDs, req.ColumnIDs, timestamps, req.Shard, opts...)
if err != nil {
api.server.logger.Errorf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
return errors.Wrap(err, "importing")
@ -1728,8 +1728,9 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
// if we're importing into a specific shard
if req.Shard != math.MaxUint64 {
// Check that column IDs match the stated shard.
if s1, s2 := req.ColumnIDs[0]/ShardWidth, req.ColumnIDs[len(req.ColumnIDs)-1]/ShardWidth; s1 != s2 && s2 != req.Shard {
return errors.Errorf("shard %d specified, but import spans shards %d to %d", req.Shard, s1, s2)
shard := req.ColumnIDs[0] / ShardWidth
if s2 := req.ColumnIDs[len(req.ColumnIDs)-1] / ShardWidth; (shard != s2) || (shard != req.Shard) {
return errors.Errorf("shard %d specified, but import spans shards %d to %d", req.Shard, shard, s2)
}
// Validate shard ownership. TODO - we should forward to the
// correct node rather than barfing here.
@ -1738,7 +1739,7 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
}
// Import columnIDs into existence field.
if !options.Clear {
if err := importExistenceColumns(qcx, idx, req.ColumnIDs); err != nil {
if err := importExistenceColumns(qcx, idx, req.ColumnIDs, shard); err != nil {
api.server.logger.Errorf("import existence error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
return errors.Wrap(err, "importing existence columns")
}
@ -1746,17 +1747,17 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
// Import into fragment.
if len(req.Values) > 0 {
err = field.importValue(qcx, req.ColumnIDs, req.Values, options)
err = field.importValue(qcx, req.ColumnIDs, req.Values, shard, options)
if err != nil {
api.server.logger.Errorf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
}
} else if len(req.TimestampValues) > 0 {
err = field.importTimestampValue(qcx, req.ColumnIDs, req.TimestampValues, options)
err = field.importTimestampValue(qcx, req.ColumnIDs, req.TimestampValues, shard, options)
if err != nil {
api.server.logger.Errorf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
}
} else if len(req.FloatValues) > 0 {
err = field.importFloatValue(qcx, req.ColumnIDs, req.FloatValues, options)
err = field.importFloatValue(qcx, req.ColumnIDs, req.FloatValues, shard, options)
if err != nil {
api.server.logger.Errorf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
}
@ -1814,14 +1815,20 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
return nil
}
func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64) error {
func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uint64) error {
ef := index.existenceField()
if ef == nil {
return nil
}
existenceRowIDs := make([]uint64, len(columnIDs))
return ef.Import(qcx, existenceRowIDs, columnIDs, nil)
// 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)
}
// ShardDistribution returns an object representing the distribution of shards

4
api_test.go
View file

@ -482,10 +482,10 @@ func TestAPI_ClearFlagForImportAndImportValues(t *testing.T) {
}
qcx := m0api.Txf().NewQcx()
if err := m0api.Import(ctx, qcx, ir0); err != nil {
if err := m0api.Import(ctx, qcx, ir0.Clone()); err != nil {
t.Fatal(err)
}
if err := m0api.ImportValue(ctx, qcx, ivr0); err != nil {
if err := m0api.ImportValue(ctx, qcx, ivr0.Clone()); err != nil {
t.Fatal(err)
}
PanicOn(qcx.Finish())

11
executor_test.go
View file

@ -4131,10 +4131,17 @@ func TestExecutor_Execute_All(t *testing.T) {
req.ColumnIDs[bitCount-1] = uint64((3 * ShardWidth) + 2)
m0 := c.GetNode(0)
// the request gets altered by the Import operation now...
reqs, err := req.Clone().ShardSplit()
if err != nil {
t.Fatalf("splitting request into shards: %v", err)
}
qcx := m0.API.Txf().NewQcx()
if err := m0.API.Import(context.Background(), qcx, req); err != nil {
t.Fatal(err)
for _, r := range reqs {
if err := m0.API.Import(context.Background(), qcx, r); err != nil {
t.Fatal(err)
}
}
PanicOn(qcx.Finish())

121
field.go
View file

@ -1437,7 +1437,7 @@ func (f *Field) Range(qcx *Qcx, name string, op pql.Token, predicate int64) (*Ro
}
// Import bulk imports data.
func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, opts ...ImportOption) (err0 error) {
func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, shard uint64, opts ...ImportOption) (err0 error) {
// Set up import options.
options := &ImportOptions{}
@ -1456,12 +1456,57 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64,
} else if options.Clear {
return errors.New("import clear is not supported with timestamps")
}
} else {
// short path: if we don't have any timestamps, we only need
// to write to exactly one view, which is always viewStandard,
// and *every* bit goes into that view, and we already verified that
// everything is in the same shard, so we can skip most of this.
fieldType := f.Type()
if fieldType == FieldTypeBool {
for _, rowID := range rowIDs {
if rowID > 1 {
return errors.New("bool field imports only support values 0 and 1")
}
}
}
tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: f.idx, Shard: shard})
if err != nil {
return errors.Wrap(err, "qcx.GetTx")
}
var err1 error
defer finisher(&err1)
view, err := f.createViewIfNotExists(viewStandard)
if err != nil {
return errors.Wrapf(err, "creating view %s", viewStandard)
}
frag, err := view.CreateFragmentIfNotExists(shard)
if err != nil {
return errors.Wrap(err, "creating fragment")
}
err1 = frag.bulkImport(tx, rowIDs, columnIDs, options)
return err1
}
fieldType := f.Type()
// Split import data by fragment.
dataByFragment := make(map[importKey]importData)
views := make(map[string]int)
var allData []importData
see := func(name string, columnID uint64, rowID uint64) {
var ok bool
var idx int
if idx, ok = views[name]; !ok {
allData = append(allData, importData{})
idx = len(allData)
views[name] = idx
}
data := allData[idx]
data.RowIDs = append(data.RowIDs, rowID)
data.ColumnIDs = append(data.ColumnIDs, columnID)
allData[idx] = data
}
for i := range rowIDs {
rowID, columnID := rowIDs[i], columnIDs[i]
@ -1472,53 +1517,41 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64,
hasTime := len(timestamps) > i && timestamps[i] != 0
var standard []string
if !hasTime {
standard = []string{viewStandard}
} else {
// Yes, we mean `0, ts`; ts is an int64 in UnixNano units,
// time.Unix takes seconds-and-nanoseconds.
standard = viewsByTime(viewStandard, time.Unix(0, timestamps[i]).UTC(), q)
if !f.options.NoStandardView {
// In order to match the logic of `SetBit()`, we want bits
// with timestamps to write to both time and standard views.
standard = append(standard, viewStandard)
// attach bit to standard view unless we have a timestamp and
// have the NoStandardView option set
if !hasTime || !f.options.NoStandardView {
see(viewStandard, columnID, rowID)
}
if hasTime {
// attach bit to all the views for this timestamp
views := viewsByTime(viewStandard, time.Unix(0, timestamps[i]).UTC(), q)
for _, view := range views {
see(view, columnID, rowID)
}
}
// Attach bit to each standard view.
for _, name := range standard {
key := importKey{View: name, Shard: columnID / ShardWidth}
data := dataByFragment[key]
data.RowIDs = append(data.RowIDs, rowID)
data.ColumnIDs = append(data.ColumnIDs, columnID)
dataByFragment[key] = data
}
}
// Import into each fragment.
for key, data := range dataByFragment {
view, err := f.createViewIfNotExists(key.View)
tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: f.idx, Shard: shard})
if err != nil {
return errors.Wrap(err, "qcx.GetTx")
}
var err1 error
defer finisher(&err1)
for viewName, idx := range views {
data := allData[idx]
view, err := f.createViewIfNotExists(viewName)
if err != nil {
return errors.Wrap(err, "creating view")
return errors.Wrapf(err, "creating view %s", viewName)
}
frag, err := view.CreateFragmentIfNotExists(key.Shard)
frag, err := view.CreateFragmentIfNotExists(shard)
if err != nil {
return errors.Wrap(err, "creating fragment")
}
tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: frag.idx, Fragment: frag, Shard: frag.shard})
if err != nil {
return errors.Wrap(err, "qcx.GetTx")
}
err1 := frag.bulkImport(tx, data.RowIDs, data.ColumnIDs, options)
err1 = frag.bulkImport(tx, data.RowIDs, data.ColumnIDs, options)
if err1 != nil {
finisher(&err1)
return err1
}
finisher(nil)
}
return nil
}
@ -1526,7 +1559,7 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64,
// 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 {
func (f *Field) importFloatValue(qcx *Qcx, columnIDs []uint64, values []float64, shard uint64, options *ImportOptions) error {
// convert values to int64 values based on scale
ivalues := make([]int64, len(values))
bsig := f.bsiGroup(f.name)
@ -1538,13 +1571,13 @@ func (f *Field) importFloatValue(qcx *Qcx, columnIDs []uint64, values []float64,
ivalues[i] = int64(fval * mult)
}
// then call importValue
return f.importValue(qcx, columnIDs, ivalues, options)
return f.importValue(qcx, columnIDs, ivalues, shard, 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 {
func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time.Time, shard uint64, options *ImportOptions) error {
ivalues := make([]int64, len(values))
bsig := f.bsiGroup(f.name)
if bsig == nil {
@ -1554,13 +1587,13 @@ func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time
for i, t := range values {
ivalues[i] = t.UnixNano() / TimeUnitNanos(f.options.TimeUnit)
}
return f.importValue(qcx, columnIDs, ivalues, options)
return f.importValue(qcx, columnIDs, ivalues, shard, options)
}
// 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) {
func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, shard uint64, options *ImportOptions) (err0 error) {
// no data to import
if len(columnIDs) == 0 {
return nil
@ -1613,9 +1646,9 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, option
}
f.mu.Unlock()
// Since all data should be for the same shard, we can just compute
// this from the first value.
shard := columnIDs[0] / ShardWidth
if columnIDs[0]/ShardWidth != shard {
return fmt.Errorf("requested import for shard %d, got record ID for shard %d", shard, columnIDs[0]/ShardWidth)
}
view, err := f.createViewIfNotExists(viewName)
if err != nil {

11
field_internal_test.go
View file

@ -494,7 +494,7 @@ func TestBSIGroup_importValue(t *testing.T) {
[]uint64{100},
},
} {
if err := f.importValue(qcx, tt.columnIDs, tt.values, options); err != nil {
if err := f.importValue(qcx, tt.columnIDs, tt.values, 0, options); err != nil {
t.Fatalf("test %d, importing values: %s", i, err.Error())
}
PanicOn(qcx.Finish())
@ -514,7 +514,8 @@ func TestBSIGroup_importValue(t *testing.T) {
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{})
// NOTE: We assume everything's in Shard 0 for now.
err := f.importValue(qcx, req.ColumnIDs, req.Values, 0, &ImportOptions{})
if err != nil {
b.Fatalf("error importing values: %s", err)
}
@ -591,7 +592,7 @@ func TestIntField_MinMaxForShard(t *testing.T) {
},
} {
t.Run(test.name+strconv.Itoa(i), func(t *testing.T) {
if err := f.importValue(qcx, test.columnIDs, test.values, options); err != nil {
if err := f.importValue(qcx, test.columnIDs, test.values, 0, options); err != nil {
t.Fatalf("test %d, importing values: %s", i, err.Error())
}
PanicOn(qcx.Finish())
@ -751,7 +752,7 @@ func TestDecimalField_MinMaxForShard(t *testing.T) {
},
} {
t.Run(test.name+strconv.Itoa(i), func(t *testing.T) {
if err := f.importFloatValue(qcx, test.columnIDs, test.values, options); err != nil {
if err := f.importFloatValue(qcx, test.columnIDs, test.values, 0, options); err != nil {
t.Fatalf("test %d, importing values: %s", i, err.Error())
}
@ -811,7 +812,7 @@ func TestBSIGroup_TxReopenDB(t *testing.T) {
[]uint64{100},
},
} {
if err := f.importValue(qcx, tt.columnIDs, tt.values, options); err != nil {
if err := f.importValue(qcx, tt.columnIDs, tt.values, 0, options); err != nil {
t.Fatalf("test %d, importing values: %s", i, err.Error())
}
PanicOn(qcx.Finish())

4
fragment_internal_test.go
View file

@ -1020,6 +1020,10 @@ func BenchmarkFragment_SetValue(b *testing.B) {
}
}
// makeBenchmarkImportValueData produces data that's supposed to be all within
// the same shard; for fragment purposes, implicitly shard 0. This also gets
// used by the field tests, but import requests are supposed to be per-shard,
// so it's important that we generate values only within a given shard.
func makeBenchmarkImportValueData(b *testing.B, bitDepth uint64, cfunc func(uint64) uint64) []ImportValueRequest {
b.StopTimer()
column := uint64(0)

118
handler.go
View file

@ -18,6 +18,7 @@ import (
"encoding/json"
"time"
"github.com/molecula/featurebase/v2/shardwidth"
"github.com/molecula/featurebase/v2/tracing"
"github.com/pkg/errors"
)
@ -129,6 +130,41 @@ type ImportValueRequest struct {
scratch []int // scratch space to allow us to get a stable sort in reasonable time
}
func (ivr *ImportValueRequest) Clone() *ImportValueRequest {
newIVR := &ImportValueRequest{}
if ivr == nil {
return newIVR
}
*newIVR = *ivr
// don't copy the internal scratch buffer
newIVR.scratch = nil
if len(ivr.ColumnIDs) > 0 {
newIVR.ColumnIDs = make([]uint64, len(ivr.ColumnIDs))
copy(newIVR.ColumnIDs, ivr.ColumnIDs)
}
if len(ivr.ColumnKeys) > 0 {
newIVR.ColumnKeys = make([]string, len(ivr.ColumnKeys))
copy(newIVR.ColumnKeys, ivr.ColumnKeys)
}
if len(ivr.Values) > 0 {
newIVR.Values = make([]int64, len(ivr.Values))
copy(newIVR.Values, ivr.Values)
}
if len(ivr.FloatValues) > 0 {
newIVR.FloatValues = make([]float64, len(ivr.FloatValues))
copy(newIVR.FloatValues, ivr.FloatValues)
}
if len(ivr.TimestampValues) > 0 {
newIVR.TimestampValues = make([]time.Time, len(ivr.TimestampValues))
copy(newIVR.TimestampValues, ivr.TimestampValues)
}
if len(ivr.StringValues) > 0 {
newIVR.StringValues = make([]string, len(ivr.StringValues))
copy(newIVR.StringValues, ivr.StringValues)
}
return newIVR
}
// AtomicRecord applies all its Ivr and Ivr atomically, in a Tx.
// The top level Shard has to agree with Ivr[i].Shard and the Iv[i].Shard
// for all i included (in Ivr and Ir). The same goes for the top level Index: all records
@ -142,6 +178,19 @@ type AtomicRecord struct {
Ir []*ImportRequest // other field types, e.g. single bit
}
func (ar *AtomicRecord) Clone() *AtomicRecord {
newAR := &AtomicRecord{Index: ar.Index, Shard: ar.Shard}
newAR.Ivr = make([]*ImportValueRequest, len(ar.Ivr))
for i, vr := range ar.Ivr {
newAR.Ivr[i] = vr.Clone()
}
newAR.Ir = make([]*ImportRequest, len(ar.Ir))
for i, vr := range ar.Ir {
newAR.Ir[i] = vr.Clone()
}
return newAR
}
func (ivr *ImportValueRequest) Len() int { return len(ivr.ColumnIDs) }
func (ivr *ImportValueRequest) Less(i, j int) bool {
if ivr.ColumnIDs[i] < ivr.ColumnIDs[j] {
@ -225,6 +274,75 @@ type ImportRequest struct {
Clear bool
}
// Clone allows copying an import request. Normally you wouldn't, but
// some import functions are destructive on their inputs, and if you
// want to *re-use* an import request, you might need this. If you're
// using this outside tx_test, something is probably wrong.
func (ir *ImportRequest) Clone() *ImportRequest {
newIR := &ImportRequest{}
if ir == nil {
return newIR
}
*newIR = *ir
if ir.RowIDs != nil {
newIR.RowIDs = make([]uint64, len(ir.RowIDs))
copy(newIR.RowIDs, ir.RowIDs)
}
if ir.ColumnIDs != nil {
newIR.ColumnIDs = make([]uint64, len(ir.ColumnIDs))
copy(newIR.ColumnIDs, ir.ColumnIDs)
}
if ir.RowKeys != nil {
newIR.RowKeys = make([]string, len(ir.RowKeys))
copy(newIR.RowKeys, ir.RowKeys)
}
if ir.ColumnKeys != nil {
newIR.ColumnKeys = make([]string, len(ir.ColumnKeys))
copy(newIR.ColumnKeys, ir.ColumnKeys)
}
if ir.Timestamps != nil {
newIR.Timestamps = make([]int64, len(ir.Timestamps))
copy(newIR.Timestamps, ir.Timestamps)
}
return newIR
}
// ShardSplit splits the request into a slice of import requests. It requires
// that the original request have all elements sorted, and already have
// column IDs, not column keys.
func (ir *ImportRequest) ShardSplit() ([]*ImportRequest, error) {
if ir == nil {
return nil, nil
}
// fix shard
if len(ir.ColumnIDs) < 2 {
ir.Shard = ir.ColumnIDs[0] >> shardwidth.Exponent
return []*ImportRequest{ir}, nil
}
shards, ends := shardwidth.FindShards(ir.ColumnIDs)
out := make([]*ImportRequest, len(shards))
prev := 0
for i, shard := range shards {
next := ends[i]
newIR := &ImportRequest{}
*newIR = *ir
newIR.ColumnIDs = ir.ColumnIDs[prev:next:next]
if ir.RowIDs != nil {
newIR.RowIDs = ir.RowIDs[prev:next:next]
}
if ir.RowKeys != nil {
newIR.RowKeys = ir.RowKeys[prev:next:next]
}
if ir.Timestamps != nil {
newIR.Timestamps = ir.Timestamps[prev:next:next]
}
newIR.Shard = shard
out[i] = newIR
prev = next
}
return out, nil
}
// ValidateWithTimestamp ensures that the payload of the request is valid.
func (ir *ImportRequest) ValidateWithTimestamp(indexCreatedAt, fieldCreatedAt int64) error {
if (ir.IndexCreatedAt != 0 && ir.IndexCreatedAt != indexCreatedAt) ||

2
http/client.go
View file

@ -30,7 +30,7 @@ import (
"strings"
"time"
"github.com/molecula/featurebase/v2"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/encoding/proto"
pnet "github.com/molecula/featurebase/v2/net"
"github.com/molecula/featurebase/v2/topology"

5
index.go
View file

@ -840,11 +840,6 @@ type IndexOptions struct {
TrackExistence bool `json:"trackExistence"`
}
type importKey struct {
View string
Shard uint64
}
type importData struct {
RowIDs []uint64
ColumnIDs []uint64

4
tx_test.go
View file

@ -203,7 +203,7 @@ func TestAPI_ImportAtomicRecord(t *testing.T) {
qcx = m0api.Txf().NewQcx()
//vv("just before the SECOND ImportAtomicRecord, qcx is %p, should NOT BE NIL", qcx)
err = m0api.ImportAtomicRecord(ctx, qcx, air, opt)
err = m0api.ImportAtomicRecord(ctx, qcx, air.Clone(), opt)
//err = m0api.ImportAtomicRecord(ctx, nil, air, opt)
if err != pilosa.ErrAborted {
PanicOn(fmt.Sprintf("expected ErrTxnAborted but got err='%#v'", err))
@ -229,7 +229,7 @@ func TestAPI_ImportAtomicRecord(t *testing.T) {
// happy path with no power failure half-way through.
qcx = m0api.Txf().NewQcx()
err = m0api.ImportAtomicRecord(ctx, qcx, air)
err = m0api.ImportAtomicRecord(ctx, qcx, air.Clone())
PanicOn(err)
if err := qcx.Finish(); err != nil {
t.Fatal(err)