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

Tx Interface

Browse source 
This commit adds a transaction interface which will be used in the
future to add support to RBF (Roaring B-tree Format).
This commit is contained in:
Ben Johnson 2020-02-28 16:24:04 -07:00
parent 549c98dc64
commit bf55bbc717
33 changed files with 2788 additions and 1619 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
api.go
View file

@ -370,7 +370,8 @@ func importWorker(importWork chan importJob) {
var doClear bool
switch doAction {
case RequestActionOverwrite:
if err := j.field.importRoaringOverwrite(j.ctx, viewData, j.shard, viewName, j.req.Block); err != nil {
tx := &RoaringTx{Field: j.field}
if err := j.field.importRoaringOverwrite(j.ctx, tx, viewData, j.shard, viewName, j.req.Block); err != nil {
return errors.Wrap(err, "importing roaring as overwrite")
}
case RequestActionClear:
@ -581,6 +582,9 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
return ErrFragmentNotFound
}
// Obtain transaction
tx := &RoaringTx{Index: index}
// Wrap writer with a CSV writer.
cw := csv.NewWriter(w)
@ -616,7 +620,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
}
// Iterate over each column.
if err := f.forEachBit(fn); err != nil {
if err := f.forEachBit(tx, fn); err != nil {
return errors.Wrap(err, "writing CSV")
}
@ -643,7 +647,7 @@ func (api *API) ShardNodes(ctx context.Context, indexName string, shard uint64)
// FragmentBlockData is an endpoint for internal usage. It is not guaranteed to
// return anything useful. Currently it returns protobuf encoded row and column
// ids from a "block" which is a subdivision of a fragment.
func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte, error) {
func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) (_ []byte, err error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.FragmentBlockData")
defer span.Finish()
@ -667,7 +671,10 @@ func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte,
}
var resp = BlockDataResponse{}
resp.RowIDs, resp.ColumnIDs = f.blockData(int(req.Block))
resp.RowIDs, resp.ColumnIDs, err = f.blockData(int(req.Block))
if err != nil {
return nil, err
}
// Encode response.
buf, err := api.Serializer.Marshal(&resp)
@ -694,8 +701,7 @@ func (api *API) FragmentBlocks(ctx context.Context, indexName, fieldName, viewNa
}
// Retrieve blocks.
blocks := f.Blocks()
return blocks, nil
return f.Blocks()
}
// FragmentData returns all data in the specified fragment.
@ -1054,6 +1060,9 @@ func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOp
return errors.Wrap(err, "getting index and field")
}
// Obtain transaction.
tx := &RoaringTx{Index: index}
if err := req.ValidateWithTimestamp(index.CreatedAt(), field.CreatedAt()); err != nil {
return errors.Wrap(err, "validating import value request")
}
@ -1146,14 +1155,14 @@ func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOp
// Import columnIDs into existence field.
if !options.Clear {
if err := importExistenceColumns(index, req.ColumnIDs); err != nil {
if err := importExistenceColumns(tx, index, req.ColumnIDs); err != nil {
api.server.logger.Printf("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")
}
}
// Import into fragment.
err = field.Import(req.RowIDs, req.ColumnIDs, timestamps, opts...)
err = field.Import(tx, req.RowIDs, req.ColumnIDs, timestamps, opts...)
if err != nil {
api.server.logger.Printf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
}
@ -1178,6 +1187,9 @@ func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts .
return errors.Wrap(err, "validating import value request")
}
// Obtain transaction.
tx := &RoaringTx{Index: index}
// Set up import options.
options, err := setUpImportOptions(opts...)
if err != nil {
@ -1244,7 +1256,7 @@ func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts .
}
// Import columnIDs into existence field.
if !options.Clear {
if err := importExistenceColumns(index, req.ColumnIDs); err != nil {
if err := importExistenceColumns(tx, index, req.ColumnIDs); err != nil {
api.server.logger.Printf("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")
}
@ -1252,12 +1264,12 @@ func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts .
// Import into fragment.
if len(req.Values) > 0 {
err = field.importValue(req.ColumnIDs, req.Values, options)
err = field.importValue(tx, req.ColumnIDs, req.Values, options)
if err != nil {
api.server.logger.Printf("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(req.ColumnIDs, req.FloatValues, options)
err = field.importFloatValue(tx, req.ColumnIDs, req.FloatValues, options)
if err != nil {
api.server.logger.Printf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
}
@ -1342,14 +1354,14 @@ func (api *API) ImportColumnAttrs(ctx context.Context, req *ImportColumnAttrsReq
return nil
}
func importExistenceColumns(index *Index, columnIDs []uint64) error {
func importExistenceColumns(tx Tx, index *Index, columnIDs []uint64) error {
ef := index.existenceField()
if ef == nil {
return nil
}
existenceRowIDs := make([]uint64, len(columnIDs))
return ef.Import(existenceRowIDs, columnIDs, nil)
return ef.Import(tx, existenceRowIDs, columnIDs, nil)
}
// MaxShards returns the maximum shard number for each index in a map.

6
cluster.go
View file

@ -755,7 +755,7 @@ func (c *cluster) fragsByHost(idx *Index) fragsByHost {
// for the given set of shards with data.
func (c *cluster) fragCombos(idx string, availableShards *roaring.Bitmap, fieldViews viewsByField) fragsByHost {
t := make(fragsByHost)
availableShards.ForEach(func(i uint64) {
_ = availableShards.ForEach(func(i uint64) error {
nodes := c.shardNodes(idx, i)
for _, n := range nodes {
// for each field/view combination:
@ -765,6 +765,7 @@ func (c *cluster) fragCombos(idx string, availableShards *roaring.Bitmap, fieldV
}
}
}
return nil
})
return t
}
@ -1060,7 +1061,7 @@ func (c *cluster) unprotectedOwnsPartition(nodeID string, partition int) bool {
// containsShards is like OwnsShards, but it includes replicas.
func (c *cluster) containsShards(index string, availableShards *roaring.Bitmap, node *Node) []uint64 {
var shards []uint64
availableShards.ForEach(func(i uint64) {
_ = availableShards.ForEach(func(i uint64) error {
p := c.shardPartition(index, i)
// Determine the nodes for partition.
nodes := c.partitionNodes(p)
@ -1069,6 +1070,7 @@ func (c *cluster) containsShards(index string, availableShards *roaring.Bitmap,
shards = append(shards, i)
}
}
return nil
})
return shards
}

23
cluster_internal_test.go
View file

@ -105,7 +105,6 @@ func newIndexWithTempPath(name string) *Index {
// Ensure that fragSources creates the correct fragment mapping.
func TestFragSources(t *testing.T) {
uri0, err := NewURIFromAddress("host0")
if err != nil {
t.Fatal(err)
@ -159,23 +158,28 @@ func TestFragSources(t *testing.T) {
idx := newIndexWithTempPath("i")
defer idx.Close()
// Obtain transaction.
tx := &RoaringTx{Index: idx}
defer func() { _ = tx.Rollback() }()
field, err := idx.CreateFieldIfNotExists("f", OptFieldTypeDefault())
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit(1, 101, nil)
_, err = field.SetBit(tx, 1, 101, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit(1, ShardWidth+1, nil)
_, err = field.SetBit(tx, 1, ShardWidth+1, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit(1, ShardWidth*2+1, nil)
_, err = field.SetBit(tx, 1, ShardWidth*2+1, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit(1, ShardWidth*3+1, nil)
_, err = field.SetBit(tx, 1, ShardWidth*3+1, nil)
if err != nil {
t.Fatal(err)
}
@ -795,7 +799,10 @@ func TestCluster_ResizeStates(t *testing.T) {
node0Field := node0.holder.Field("i", "f")
node0View := node0Field.view("standard")
node0Fragment := node0View.Fragment(1)
node0Checksum := node0Fragment.Checksum()
node0Checksum, err := node0Fragment.Checksum()
if err != nil {
t.Fatal(err)
}
// addNode needs to block until the resize process has completed.
if err := tc.addNode(); err != nil {
@ -828,7 +835,9 @@ func TestCluster_ResizeStates(t *testing.T) {
node1Fragment := node1View.Fragment(1)
// Ensure checksums are the same.
if chksum := node1Fragment.Checksum(); !bytes.Equal(chksum, node0Checksum) {
if chksum, err := node1Fragment.Checksum(); err != nil {
t.Fatal(err)
} else if !bytes.Equal(chksum, node0Checksum) {
t.Fatalf("expected standard view checksum to match: %x - %x", chksum, node0Checksum)
}

0
cmd/inspect.go → cmd/convert.go
View file

415
executor.go
View file

File diff suppressed because it is too large Load diff

19
executor_internal_test.go
View file

@ -137,13 +137,18 @@ func TestExecutor_TranslateRowsOnBool(t *testing.T) {
holder := NewHolder(DefaultPartitionN)
defer holder.Close()
tx, err := holder.Begin(true)
if err != nil {
t.Fatal(err)
}
defer func() { _ = tx.Rollback() }()
e := &executor{
Holder: holder,
Cluster: NewTestCluster(1),
}
e.Holder.Path, _ = ioutil.TempDir(*TempDir, "")
err := e.Holder.Open()
if err != nil {
if err := e.Holder.Open(); err != nil {
t.Fatalf("opening holder: %v", err)
}
@ -158,13 +163,17 @@ func TestExecutor_TranslateRowsOnBool(t *testing.T) {
t.Fatalf("creating fields %v, %v", errb, errbk)
}
_, err1 := fb.SetBit(1, 1, nil)
_, err2 := fb.SetBit(2, 2, nil)
_, err3 := fb.SetBit(3, 3, nil)
_, err1 := fb.SetBit(tx, 1, 1, nil)
_, err2 := fb.SetBit(tx, 2, 2, nil)
_, err3 := fb.SetBit(tx, 3, 3, nil)
if err1 != nil || err2 != nil || err3 != nil {
t.Fatalf("setting bit %v, %v, %v", err1, err2, err3)
}
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
tests := []struct {
pql string
}{

11
executor_test.go
View file

@ -903,8 +903,15 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
t.Fatal(err)
}
// Obtain transaction.
tx, err := hldr.Begin(false)
if err != nil {
t.Fatal(err)
}
defer func() { _ = tx.Rollback() }()
f := hldr.Field("i", "f")
if value, exists, err := f.Value(10); err != nil {
if value, exists, err := f.Value(tx, 10); err != nil {
t.Fatal(err)
} else if !exists {
t.Fatal("expected value to exist")
@ -912,7 +919,7 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
t.Fatalf("unexpected value: %v", value)
}
if value, exists, err := f.Value(100); err != nil {
if value, exists, err := f.Value(tx, 100); err != nil {
t.Fatal(err)
} else if !exists {
t.Fatal("expected value to exist")

75
field.go
View file

@ -1088,7 +1088,7 @@ func (f *Field) setTimeQuantum(q TimeQuantum) error {
// RowTime gets the row at the particular time with the granularity specified by
// the quantum.
func (f *Field) RowTime(rowID uint64, time time.Time, quantum string) (*Row, error) {
func (f *Field) RowTime(tx Tx, rowID uint64, time time.Time, quantum string) (*Row, error) {
if !TimeQuantum(quantum).Valid() {
return nil, ErrInvalidTimeQuantum
}
@ -1097,7 +1097,7 @@ func (f *Field) RowTime(rowID uint64, time time.Time, quantum string) (*Row, err
if view == nil {
return nil, errors.Errorf("view with quantum %v not found.", quantum)
}
return view.row(rowID), nil
return view.row(tx, rowID)
}
// viewPath returns the path to a view in the field.
@ -1212,21 +1212,21 @@ func (f *Field) deleteView(name string) error {
// package, and the fact that it's only allowed on
// `set`,`mutex`, and `bool` fields is odd. This may
// be considered for deprecation in a future version.
func (f *Field) Row(rowID uint64) (*Row, error) {
func (f *Field) Row(tx Tx, rowID uint64) (*Row, error) {
switch f.Type() {
case FieldTypeSet, FieldTypeMutex, FieldTypeBool:
view := f.view(viewStandard)
if view == nil {
return nil, ErrInvalidView
}
return view.row(rowID), nil
return view.row(tx, rowID)
default:
return nil, errors.Errorf("row method unsupported for field type: %s", f.Type())
}
}
// SetBit sets a bit on a view within the field.
func (f *Field) SetBit(rowID, colID uint64, t *time.Time) (changed bool, err error) {
func (f *Field) SetBit(tx Tx, rowID, colID uint64, t *time.Time) (changed bool, err error) {
viewName := viewStandard
if !f.options.NoStandardView {
// Retrieve view. Exit if it doesn't exist.
@ -1236,7 +1236,7 @@ func (f *Field) SetBit(rowID, colID uint64, t *time.Time) (changed bool, err err
}
// Set non-time bit.
if v, err := view.setBit(rowID, colID); err != nil {
if v, err := view.setBit(tx, rowID, colID); err != nil {
return changed, errors.Wrap(err, "setting on view")
} else if v {
changed = v
@ -1255,7 +1255,7 @@ func (f *Field) SetBit(rowID, colID uint64, t *time.Time) (changed bool, err err
return changed, errors.Wrapf(err, "creating view %s", subname)
}
if c, err := view.setBit(rowID, colID); err != nil {
if c, err := view.setBit(tx, rowID, colID); err != nil {
return changed, errors.Wrapf(err, "setting on view %s", subname)
} else if c {
changed = true
@ -1266,7 +1266,7 @@ func (f *Field) SetBit(rowID, colID uint64, t *time.Time) (changed bool, err err
}
// ClearBit clears a bit within the field.
func (f *Field) ClearBit(rowID, colID uint64) (changed bool, err error) {
func (f *Field) ClearBit(tx Tx, rowID, colID uint64) (changed bool, err error) {
viewName := viewStandard
// Retrieve view. Exit if it doesn't exist.
@ -1276,7 +1276,7 @@ func (f *Field) ClearBit(rowID, colID uint64) (changed bool, err error) {
}
// Clear non-time bit.
if v, err := view.clearBit(rowID, colID); err != nil {
if v, err := view.clearBit(tx, rowID, colID); err != nil {
return false, errors.Wrap(err, "clearing on view")
} else if v {
changed = changed || v
@ -1294,7 +1294,7 @@ func (f *Field) ClearBit(rowID, colID uint64) (changed bool, err error) {
level--
}
if level < skipAbove {
cleared, err := view.clearBit(rowID, colID)
cleared, err := view.clearBit(tx, rowID, colID)
changed = changed || cleared
if err != nil {
return changed, errors.Wrapf(err, "clearing on view %s", view.name)
@ -1354,13 +1354,13 @@ func (f *Field) allTimeViewsSortedByQuantum() (me []*view) {
// StringValue reads an integer field value for a column, and converts
// it to a string based on a foreign index string key.
func (f *Field) StringValue(columnID uint64) (value string, exists bool, err error) {
func (f *Field) StringValue(tx Tx, columnID uint64) (value string, exists bool, err error) {
bsig := f.bsiGroup(f.name)
if bsig == nil {
return value, false, ErrBSIGroupNotFound
}
val, exists, err := f.Value(columnID)
val, exists, err := f.Value(tx, columnID)
if exists {
value, err = f.translateStore.TranslateID(uint64(val))
}
@ -1368,7 +1368,7 @@ func (f *Field) StringValue(columnID uint64) (value string, exists bool, err err
}
// Value reads a field value for a column.
func (f *Field) Value(columnID uint64) (value int64, exists bool, err error) {
func (f *Field) Value(tx Tx, columnID uint64) (value int64, exists bool, err error) {
bsig := f.bsiGroup(f.name)
if bsig == nil {
return 0, false, ErrBSIGroupNotFound
@ -1380,7 +1380,7 @@ func (f *Field) Value(columnID uint64) (value int64, exists bool, err error) {
return 0, false, nil
}
v, exists, err := view.value(columnID, bsig.BitDepth)
v, exists, err := view.value(tx, columnID, bsig.BitDepth)
if err != nil {
return 0, false, err
} else if !exists {
@ -1390,7 +1390,7 @@ func (f *Field) Value(columnID uint64) (value int64, exists bool, err error) {
}
// SetValue sets a field value for a column.
func (f *Field) SetValue(columnID uint64, value int64) (changed bool, err error) {
func (f *Field) SetValue(tx Tx, columnID uint64, value int64) (changed bool, err error) {
// Fetch bsiGroup & validate min/max.
bsig := f.bsiGroup(f.name)
if bsig == nil {
@ -1430,11 +1430,11 @@ func (f *Field) SetValue(columnID uint64, value int64) (changed bool, err error)
if err != nil {
return false, errors.Wrap(err, "creating view")
}
return view.setValue(columnID, bsig.BitDepth, baseValue)
return view.setValue(tx, columnID, bsig.BitDepth, baseValue)
}
// ClearValue removes a field value for a column.
func (f *Field) ClearValue(columnID uint64) (changed bool, err error) {
func (f *Field) ClearValue(tx Tx, columnID uint64) (changed bool, err error) {
bsig := f.bsiGroup(f.name)
if bsig == nil {
return false, ErrBSIGroupNotFound
@ -1444,17 +1444,17 @@ func (f *Field) ClearValue(columnID uint64) (changed bool, err error) {
if view == nil {
return false, nil
}
value, exists, err := view.value(columnID, bsig.BitDepth)
value, exists, err := view.value(tx, columnID, bsig.BitDepth)
if err != nil {
return false, err
}
if exists {
return view.clearValue(columnID, bsig.BitDepth, value)
return view.clearValue(tx, columnID, bsig.BitDepth, value)
}
return false, nil
}
func (f *Field) MaxForShard(shard uint64, filter *Row) (ValCount, error) {
func (f *Field) MaxForShard(tx Tx, shard uint64, filter *Row) (ValCount, error) {
bsig := f.bsiGroup(f.name)
if bsig == nil {
return ValCount{}, ErrBSIGroupNotFound
@ -1470,7 +1470,7 @@ func (f *Field) MaxForShard(shard uint64, filter *Row) (ValCount, error) {
return ValCount{}, nil
}
max, cnt, err := fragment.max(filter, bsig.BitDepth)
max, cnt, err := fragment.max(tx, filter, bsig.BitDepth)
if err != nil {
return ValCount{}, errors.Wrap(err, "calling fragment.max")
}
@ -1490,7 +1490,7 @@ func (f *Field) MaxForShard(shard uint64, filter *Row) (ValCount, error) {
// MinForShard returns the minimum value which appears in this shard
// (this field must be an Int or Decimal field). It also returns the
// number of times the minimum value appears.
func (f *Field) MinForShard(shard uint64, filter *Row) (ValCount, error) {
func (f *Field) MinForShard(tx Tx, shard uint64, filter *Row) (ValCount, error) {
bsig := f.bsiGroup(f.name)
if bsig == nil {
return ValCount{}, ErrBSIGroupNotFound
@ -1506,7 +1506,7 @@ func (f *Field) MinForShard(shard uint64, filter *Row) (ValCount, error) {
return ValCount{}, nil
}
min, cnt, err := fragment.min(filter, bsig.BitDepth)
min, cnt, err := fragment.min(tx, filter, bsig.BitDepth)
if err != nil {
return ValCount{}, errors.Wrap(err, "calling fragment.min")
}
@ -1524,7 +1524,7 @@ func (f *Field) MinForShard(shard uint64, filter *Row) (ValCount, error) {
}
// Range performs a conditional operation on Field.
func (f *Field) Range(name string, op pql.Token, predicate int64) (*Row, error) {
func (f *Field) Range(tx Tx, name string, op pql.Token, predicate int64) (*Row, error) {
// Retrieve and validate bsiGroup.
bsig := f.bsiGroup(name)
if bsig == nil {
@ -1544,11 +1544,11 @@ func (f *Field) Range(name string, op pql.Token, predicate int64) (*Row, error)
return NewRow(), nil
}
return view.rangeOp(op, bsig.BitDepth, baseValue)
return view.rangeOp(tx, op, bsig.BitDepth, baseValue)
}
// Import bulk imports data.
func (f *Field) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time, opts ...ImportOption) error {
func (f *Field) Import(tx Tx, rowIDs, columnIDs []uint64, timestamps []*time.Time, opts ...ImportOption) error {
// Set up import options.
options := &ImportOptions{}
@ -1620,7 +1620,7 @@ func (f *Field) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time, opts
return errors.Wrap(err, "creating fragment")
}
if err := frag.bulkImport(data.RowIDs, data.ColumnIDs, options); err != nil {
if err := frag.bulkImport(tx, data.RowIDs, data.ColumnIDs, options); err != nil {
return err
}
}
@ -1628,7 +1628,7 @@ func (f *Field) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time, opts
return nil
}
func (f *Field) importFloatValue(columnIDs []uint64, values []float64, options *ImportOptions) error {
func (f *Field) importFloatValue(tx Tx, columnIDs []uint64, values []float64, options *ImportOptions) error {
// convert values to int64 values based on scale
ivalues := make([]int64, len(values))
bsig := f.bsiGroup(f.name)
@ -1640,11 +1640,11 @@ func (f *Field) importFloatValue(columnIDs []uint64, values []float64, options *
ivalues[i] = int64(fval * mult)
}
// then call importValue
return f.importValue(columnIDs, ivalues, options)
return f.importValue(tx, columnIDs, ivalues, options)
}
// importValue bulk imports range-encoded value data.
func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportOptions) error {
func (f *Field) importValue(tx Tx, columnIDs []uint64, values []int64, options *ImportOptions) error {
viewName := viewBSIGroupPrefix + f.name
// Get the bsiGroup so we know bitDepth.
bsig := f.bsiGroup(f.name)
@ -1727,7 +1727,7 @@ func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportO
baseValues[i] = value - bsig.Base
}
if err := frag.importValue(data.ColumnIDs, baseValues, requiredDepth, options.Clear); err != nil {
if err := frag.importValue(tx, data.ColumnIDs, baseValues, requiredDepth, options.Clear); err != nil {
return err
}
}
@ -1759,7 +1759,7 @@ func (f *Field) importRoaring(ctx context.Context, data []byte, shard uint64, vi
return nil
}
func (f *Field) importRoaringOverwrite(ctx context.Context, data []byte, shard uint64, viewName string, block int) error {
func (f *Field) importRoaringOverwrite(ctx context.Context, tx Tx, data []byte, shard uint64, viewName string, block int) error {
span, ctx := tracing.StartSpanFromContext(ctx, "Field.importRoaringOverwrite")
defer span.Finish()
@ -1776,7 +1776,7 @@ func (f *Field) importRoaringOverwrite(ctx context.Context, data []byte, shard u
if err != nil {
return errors.Wrap(err, "creating fragment")
}
if err := frag.importRoaringOverwrite(ctx, data, block); err != nil {
if err := frag.importRoaringOverwrite(ctx, tx, data, block); err != nil {
return err
}
@ -1785,9 +1785,14 @@ func (f *Field) importRoaringOverwrite(ctx context.Context, data []byte, shard u
switch f.Options().Type {
case FieldTypeInt, FieldTypeDecimal:
frag.mu.Lock()
frag.calculateMaxRowID()
maxRowID, _ := frag.maxRow(nil)
if err := frag.calculateMaxRowID(); err != nil {
return err
}
maxRowID, _, err := frag.maxRow(tx, nil)
frag.mu.Unlock()
if err != nil {
return err
}
var bitDepth uint
if maxRowID+1 > bsiOffsetBit {

51
field_internal_test.go
View file

@ -246,15 +246,15 @@ func (f *TestField) Reopen() error {
return nil
}
func (f *TestField) MustSetBit(row, col uint64, ts ...time.Time) {
func (f *TestField) MustSetBit(tx Tx, row, col uint64, ts ...time.Time) {
if len(ts) == 0 {
_, err := f.Field.SetBit(row, col, nil)
_, err := f.Field.SetBit(tx, row, col, nil)
if err != nil {
panic(err)
}
}
for _, t := range ts {
_, err := f.Field.SetBit(row, col, &t)
_, err := f.Field.SetBit(tx, row, col, &t)
if err != nil {
panic(err)
}
@ -310,41 +310,44 @@ func TestField_RowTime(t *testing.T) {
f := OpenField(t, OptFieldTypeTime(TimeQuantum("")))
defer f.Close()
// Obtain transaction.
tx := &RoaringTx{Field: f.Field}
if err := f.setTimeQuantum(TimeQuantum("YMDH")); err != nil {
t.Fatal(err)
}
f.MustSetBit(1, 1, time.Date(2010, time.January, 5, 12, 0, 0, 0, time.UTC))
f.MustSetBit(1, 2, time.Date(2011, time.January, 5, 12, 0, 0, 0, time.UTC))
f.MustSetBit(1, 3, time.Date(2010, time.February, 5, 12, 0, 0, 0, time.UTC))
f.MustSetBit(1, 4, time.Date(2010, time.January, 6, 12, 0, 0, 0, time.UTC))
f.MustSetBit(1, 5, time.Date(2010, time.January, 5, 13, 0, 0, 0, time.UTC))
f.MustSetBit(tx, 1, 1, time.Date(2010, time.January, 5, 12, 0, 0, 0, time.UTC))
f.MustSetBit(tx, 1, 2, time.Date(2011, time.January, 5, 12, 0, 0, 0, time.UTC))
f.MustSetBit(tx, 1, 3, time.Date(2010, time.February, 5, 12, 0, 0, 0, time.UTC))
f.MustSetBit(tx, 1, 4, time.Date(2010, time.January, 6, 12, 0, 0, 0, time.UTC))
f.MustSetBit(tx, 1, 5, time.Date(2010, time.January, 5, 13, 0, 0, 0, time.UTC))
if r, err := f.RowTime(1, time.Date(2010, time.November, 5, 12, 0, 0, 0, time.UTC), "Y"); err != nil {
if r, err := f.RowTime(tx, 1, time.Date(2010, time.November, 5, 12, 0, 0, 0, time.UTC), "Y"); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(r.Columns(), []uint64{1, 3, 4, 5}) {
t.Fatalf("wrong columns: %#v", r.Columns())
}
if r, err := f.RowTime(1, time.Date(2010, time.February, 7, 13, 0, 0, 0, time.UTC), "YM"); err != nil {
if r, err := f.RowTime(tx, 1, time.Date(2010, time.February, 7, 13, 0, 0, 0, time.UTC), "YM"); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(r.Columns(), []uint64{3}) {
t.Fatalf("wrong columns: %#v", r.Columns())
}
if r, err := f.RowTime(1, time.Date(2010, time.February, 7, 13, 0, 0, 0, time.UTC), "M"); err != nil {
if r, err := f.RowTime(tx, 1, time.Date(2010, time.February, 7, 13, 0, 0, 0, time.UTC), "M"); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(r.Columns(), []uint64{3}) {
t.Fatalf("wrong columns: %#v", r.Columns())
}
if r, err := f.RowTime(1, time.Date(2010, time.January, 5, 12, 0, 0, 0, time.UTC), "MD"); err != nil {
if r, err := f.RowTime(tx, 1, time.Date(2010, time.January, 5, 12, 0, 0, 0, time.UTC), "MD"); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(r.Columns(), []uint64{1, 5}) {
t.Fatalf("wrong columns: %#v", r.Columns())
}
if r, err := f.RowTime(1, time.Date(2010, time.January, 5, 13, 0, 0, 0, time.UTC), "MDH"); err != nil {
if r, err := f.RowTime(tx, 1, time.Date(2010, time.January, 5, 13, 0, 0, 0, time.UTC), "MDH"); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(r.Columns(), []uint64{5}) {
t.Fatalf("wrong columns: %#v", r.Columns())
@ -578,11 +581,13 @@ func TestBSIGroup_importValue(t *testing.T) {
[]uint64{100},
},
} {
if err := f.importValue(tt.columnIDs, tt.values, options); err != nil {
tx := &RoaringTx{Field: f.Field}
if err := f.importValue(tx, tt.columnIDs, tt.values, options); err != nil {
t.Fatalf("test %d, importing values: %s", i, err.Error())
}
if row, err := f.Range(f.name, pql.EQ, tt.checkVal); err != nil {
if row, err := f.Range(tx, f.name, pql.EQ, tt.checkVal); err != nil {
t.Fatalf("test %d, getting range: %s", i, err.Error())
} else if !reflect.DeepEqual(row.Columns(), tt.expCols) {
t.Fatalf("test %d, expected columns: %v, but got: %v", i, tt.expCols, row.Columns())
@ -643,11 +648,13 @@ func TestIntField_MinMaxForShard(t *testing.T) {
},
} {
t.Run(test.name+strconv.Itoa(i), func(t *testing.T) {
if err := f.importValue(test.columnIDs, test.values, options); err != nil {
tx := &RoaringTx{Field: f.Field}
if err := f.importValue(tx, test.columnIDs, test.values, options); err != nil {
t.Fatalf("test %d, importing values: %s", i, err.Error())
}
maxvc, err := f.MaxForShard(0, nil)
maxvc, err := f.MaxForShard(tx, 0, nil)
if err != nil {
t.Fatalf("getting max for shard: %v", err)
}
@ -655,7 +662,7 @@ func TestIntField_MinMaxForShard(t *testing.T) {
t.Fatalf("max expected:\n%+v\ngot:\n%+v", test.expMax, maxvc)
}
minvc, err := f.MinForShard(0, nil)
minvc, err := f.MinForShard(tx, 0, nil)
if err != nil {
t.Fatalf("getting min for shard: %v", err)
}
@ -797,11 +804,13 @@ func TestDecimalField_MinMaxForShard(t *testing.T) {
},
} {
t.Run(test.name+strconv.Itoa(i), func(t *testing.T) {
if err := f.importFloatValue(test.columnIDs, test.values, options); err != nil {
tx := &RoaringTx{Field: f.Field}
if err := f.importFloatValue(tx, test.columnIDs, test.values, options); err != nil {
t.Fatalf("test %d, importing values: %s", i, err.Error())
}
maxvc, err := f.MaxForShard(0, nil)
maxvc, err := f.MaxForShard(tx, 0, nil)
if err != nil {
t.Fatalf("getting max for shard: %v", err)
}
@ -809,7 +818,7 @@ func TestDecimalField_MinMaxForShard(t *testing.T) {
t.Fatalf("max expected:\n%+v\ngot:\n%+v", test.expMax, maxvc)
}
minvc, err := f.MinForShard(0, nil)
minvc, err := f.MinForShard(tx, 0, nil)
if err != nil {
t.Fatalf("getting min for shard: %v", err)
}

37
field_test.go
View file

@ -35,16 +35,17 @@ func TestField_SetValue(t *testing.T) {
if err != nil {
t.Fatal(err)
}
tx := &pilosa.RoaringTx{Field: f.Field}
// Set value on field.
if changed, err := f.SetValue(100, 21); err != nil {
if changed, err := f.SetValue(tx, 100, 21); err != nil {
t.Fatal(err)
} else if !changed {
t.Fatal("expected change")
}
// Read value.
if value, exists, err := f.Value(100); err != nil {
if value, exists, err := f.Value(tx, 100); err != nil {
t.Fatal(err)
} else if value != 21 {
t.Fatalf("unexpected value: %d", value)
@ -53,7 +54,7 @@ func TestField_SetValue(t *testing.T) {
}
// Setting value should return no change.
if changed, err := f.SetValue(100, 21); err != nil {
if changed, err := f.SetValue(tx, 100, 21); err != nil {
t.Fatal(err)
} else if changed {
t.Fatal("expected no change")
@ -68,23 +69,24 @@ func TestField_SetValue(t *testing.T) {
if err != nil {
t.Fatal(err)
}
tx := &pilosa.RoaringTx{Field: f.Field}
// Set value.
if changed, err := f.SetValue(100, 21); err != nil {
if changed, err := f.SetValue(tx, 100, 21); err != nil {
t.Fatal(err)
} else if !changed {
t.Fatal("expected change")
}
// Set different value.
if changed, err := f.SetValue(100, 23); err != nil {
if changed, err := f.SetValue(tx, 100, 23); err != nil {
t.Fatal(err)
} else if !changed {
t.Fatal("expected change")
}
// Read value.
if value, exists, err := f.Value(100); err != nil {
if value, exists, err := f.Value(tx, 100); err != nil {
t.Fatal(err)
} else if value != 23 {
t.Fatalf("unexpected value: %d", value)
@ -101,9 +103,10 @@ func TestField_SetValue(t *testing.T) {
if err != nil {
t.Fatal(err)
}
tx := &pilosa.RoaringTx{Field: f.Field}
// Set value.
if _, err := f.SetValue(100, 21); err != pilosa.ErrBSIGroupNotFound {
if _, err := f.SetValue(tx, 100, 21); err != pilosa.ErrBSIGroupNotFound {
t.Fatalf("unexpected error: %s", err)
}
})
@ -116,9 +119,10 @@ func TestField_SetValue(t *testing.T) {
if err != nil {
t.Fatal(err)
}
tx := &pilosa.RoaringTx{Field: f.Field}
// Set value.
if _, err := f.SetValue(100, 15); err != pilosa.ErrBSIGroupValueTooLow {
if _, err := f.SetValue(tx, 100, 15); err != pilosa.ErrBSIGroupValueTooLow {
t.Fatalf("unexpected error: %s", err)
}
})
@ -131,9 +135,10 @@ func TestField_SetValue(t *testing.T) {
if err != nil {
t.Fatal(err)
}
tx := &pilosa.RoaringTx{Field: f.Field}
// Set value.
if _, err := f.SetValue(100, 31); err != pilosa.ErrBSIGroupValueTooHigh {
if _, err := f.SetValue(tx, 100, 31); err != pilosa.ErrBSIGroupValueTooHigh {
t.Fatalf("unexpected error: %s", err)
}
})
@ -199,11 +204,12 @@ func TestField_AvailableShards(t *testing.T) {
if err != nil {
t.Fatal(err)
}
tx := &pilosa.RoaringTx{Field: f.Field}
// Set values on shards 0 & 2, and verify.
if _, err := f.SetBit(0, 100, nil); err != nil {
if _, err := f.SetBit(tx, 0, 100, nil); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(0, ShardWidth*2, nil); err != nil {
} else if _, err := f.SetBit(tx, 0, ShardWidth*2, nil); err != nil {
t.Fatal(err)
} else if diff := cmp.Diff(f.AvailableShards().Slice(), []uint64{0, 2}); diff != "" {
t.Fatal(diff)
@ -238,16 +244,17 @@ func TestField_ClearValue(t *testing.T) {
if err != nil {
t.Fatal(err)
}
tx := &pilosa.RoaringTx{Field: f.Field}
// Set value on field.
if changed, err := f.SetValue(100, 21); err != nil {
if changed, err := f.SetValue(tx, 100, 21); err != nil {
t.Fatal(err)
} else if !changed {
t.Fatal("expected change")
}
// Read value.
if value, exists, err := f.Value(100); err != nil {
if value, exists, err := f.Value(tx, 100); err != nil {
t.Fatal(err)
} else if value != 21 {
t.Fatalf("unexpected value: %d", value)
@ -255,14 +262,14 @@ func TestField_ClearValue(t *testing.T) {
t.Fatal("expected value to exist")
}
if changed, err := f.ClearValue(100); err != nil {
if changed, err := f.ClearValue(tx, 100); err != nil {
t.Fatal(err)
} else if !changed {
t.Fatal(err)
}
// Read value.
if _, exists, err := f.Value(100); err != nil {
if _, exists, err := f.Value(tx, 100); err != nil {
t.Fatal(err)
} else if exists {
t.Fatal("expected value to not exist")

708
fragment.go
View file

File diff suppressed because it is too large Load diff

847
fragment_internal_test.go
View file

File diff suppressed because it is too large Load diff

2
go.mod
View file

@ -6,8 +6,10 @@ require (
github.com/CAFxX/gcnotifier v0.0.0-20190112062741-224a280d589d
github.com/DataDog/datadog-go v0.0.0-20180822151419-281ae9f2d895
github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d // indirect
github.com/benbjohnson/immutable v0.2.0
github.com/boltdb/bolt v1.3.1
github.com/cespare/xxhash v1.1.0
github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e
github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd // indirect
github.com/davecgh/go-spew v1.1.1
github.com/go-ole/go-ole v1.2.4 // indirect

6
go.sum
View file

@ -14,6 +14,8 @@ github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRF
github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da h1:8GUt8eRujhVEGZFFEjBj46YV4rDjvGrNxb0KMWYkL2I=
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
github.com/benbjohnson/immutable v0.2.0 h1:t0rW3lNFwfQ85IDO1mhMbumxdVSti4nnVaal4r45Oio=
github.com/benbjohnson/immutable v0.2.0/go.mod h1:uc6OHo6PN2++n98KHLxW8ef4W42ylHiQSENghE1ezxI=
github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
github.com/beorn7/perks v1.0.0 h1:HWo1m869IqiPhD389kmkxeTalrjNbbJTC8LXupb+sl0=
github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
@ -22,6 +24,8 @@ github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx2
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=
github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e h1:fY5BOSpyZCqRo5OhCuC+XN+r/bBCmeuuJtjz+bCNIf8=
github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd h1:qMd81Ts1T2OTKmB4acZcyKaMtRnY5Y44NuXGX2GFJ1w=
@ -112,6 +116,7 @@ github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181
github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021 h1:ERLyN4p3KS5Fk2ADsDENm2cq0+Lx6sF1sG8uwRlySpU=
github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
github.com/pilosa/pilosa v1.4.0 h1:nqHNIK4nDslFnem3yDp9R+6TgLdlkY9WdJD88Z83T8U=
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@ -242,3 +247,4 @@ modernc.org/mathutil v1.0.0 h1:93vKjrJopTPrtTNpZ8XIovER7iCIH1QU7wNbOQXC60I=
modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=
modernc.org/strutil v1.0.0 h1:XVFtQwFVwc02Wk+0L/Z/zDDXO81r5Lhe6iMKmGX3KhE=
modernc.org/strutil v1.0.0/go.mod h1:lstksw84oURvj9y3tn8lGvRxyRC1S2+g5uuIzNfIOBs=
vitess.io/vitess v2.1.1+incompatible h1:nuuGHiWYWpudD3gOCLeGzol2EJ25e/u5Wer2wV1O130=

5
holder.go
View file

@ -613,6 +613,11 @@ func (h *Holder) Close() error {
return nil
}
// Begin starts a transaction on the holder.
func (h *Holder) Begin(writable bool) (Tx, error) {
return NewMultiTx(writable, h), nil
}
// HasData returns true if Holder contains at least one index.
// This is used to determine if the rebalancing of data is necessary
// when a node joins the cluster.

11
holder_internal_test.go
View file

@ -83,6 +83,12 @@ func makeHolder() (*Holder, string, error) {
}
func testSetBit(t *testing.T, h *Holder, index, field string, rowID, columnID uint64) {
tx, err := h.Begin(true)
if err != nil {
t.Fatal(err)
}
defer func() { _ = tx.Rollback() }()
idx, err := h.CreateIndexIfNotExists(index, IndexOptions{})
if err != nil {
t.Fatalf("creating index: %v", err)
@ -91,10 +97,13 @@ func testSetBit(t *testing.T, h *Holder, index, field string, rowID, columnID ui
if err != nil {
t.Fatalf("setting bit: %v", err)
}
_, err = f.SetBit(rowID, columnID, nil)
_, err = f.SetBit(tx, rowID, columnID, nil)
if err != nil {
t.Fatalf("setting bit: %v", err)
}
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}
func TestHolderOperatorProcess(t *testing.T) {

30
holder_test.go
View file

@ -162,11 +162,19 @@ func TestHolder_Open(t *testing.T) {
h := test.MustOpenHolder()
defer h.Close()
tx, err := h.Begin(true)
if err != nil {
t.Fatal(err)
}
defer func() { _ = tx.Rollback() }()
if idx, err := h.CreateIndex("foo", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
} else if field, err := idx.CreateField("bar", pilosa.OptFieldTypeDefault()); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(0, 0, nil); err != nil {
} else if _, err := field.SetBit(tx, 0, 0, nil); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
@ -184,11 +192,19 @@ func TestHolder_Open(t *testing.T) {
h := test.MustOpenHolder()
defer h.Close()
tx, err := h.Begin(true)
if err != nil {
t.Fatal(err)
}
defer func() { _ = tx.Rollback() }()
if idx, err := h.CreateIndex("foo", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
} else if field, err := idx.CreateField("bar", pilosa.OptFieldTypeDefault()); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(0, 0, nil); err != nil {
} else if _, err := field.SetBit(tx, 0, 0, nil); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
@ -204,11 +220,19 @@ func TestHolder_Open(t *testing.T) {
h := test.MustOpenHolder()
defer h.Close()
tx, err := h.Begin(true)
if err != nil {
t.Fatal(err)
}
defer func() { _ = tx.Rollback() }()
if idx, err := h.CreateIndex("foo", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
} else if field, err := idx.CreateField("bar", pilosa.OptFieldTypeDefault()); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(0, 0, nil); err != nil {
} else if _, err := field.SetBit(tx, 0, 0, nil); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)

6
index.go
View file

@ -362,6 +362,12 @@ func (i *Index) AvailableShards() *roaring.Bitmap {
return b
}
// Begin starts a transaction on a shard of the index.
func (i *Index) Begin(writable bool, shard uint64) (Tx, error) {
// TODO(bbj): Check for underlying storage as RBF or roaring.
return &RoaringTx{Index: i}, nil
}
// fieldPath returns the path to a field in the index.
func (i *Index) fieldPath(name string) string { return filepath.Join(i.path, name) }

6
mmap_test.go
View file

@ -35,8 +35,10 @@ func forceSnapshotsCheckMapping(t *testing.T) {
f.Logger = logger.NewLogfLogger(t)
defer f.Clean(t)
tx := &RoaringTx{fragment: f}
for i := 0; i < f.MaxOpN; i++ {
_, _ = f.setBit(0, uint64(i*32))
_, _ = f.setBit(tx, 0, uint64(32*i))
}
// force snapshot so we get a mmapped row...
err := f.Snapshot()
@ -67,7 +69,7 @@ func forceSnapshotsCheckMapping(t *testing.T) {
if i%5 == 0 {
runtime.GC()
}
err := f.importValue(cv.cols, cv.vals, depth, (i%3 == 1))
err := f.importValue(tx, cv.cols, cv.vals, depth, (i%3 == 1))
if err != nil {
t.Fatalf("importValue[%d]: %v", i, err)
}

4
roaring/btree_test.go
View file

@ -399,6 +399,7 @@ func BenchmarkBtreeSetSeq1e6(b *testing.B) {
func benchmarkSetSeq(b *testing.B, n int) {
b.ResetTimer()
b.ReportAllocs()
for i := 0; i < b.N; i++ {
b.StopTimer()
r := treeNew()
@ -436,6 +437,7 @@ func benchmarkGetSeq(b *testing.B, n int) {
}
debug.FreeOSMemory()
b.ResetTimer()
b.ReportAllocs()
for i := 0; i < b.N; i++ {
for j := 0; j < n; j++ {
r.Get(uint64(j))
@ -468,6 +470,7 @@ func benchmarkSetRnd(b *testing.B, n int) {
a[i] = rng.Next()
}
b.ResetTimer()
b.ReportAllocs()
c := getDummyC(1)
for i := 0; i < b.N; i++ {
b.StopTimer()
@ -512,6 +515,7 @@ func benchmarkGetRnd(b *testing.B, n int) {
}
debug.FreeOSMemory()
b.ResetTimer()
b.ReportAllocs()
for i := 0; i < b.N; i++ {
for _, v := range a {
r.Get(uint64(v))

52
roaring/container_stash.go
View file

@ -165,29 +165,29 @@ func NewContainerArrayN(set []uint16, n int32) *Container {
// NewContainerRun creates a new run container using a provided (possibly nil)
// slice of intervals.
func NewContainerRun(set []interval16) *Container {
func NewContainerRun(set []Interval16) *Container {
c := &Container{typeID: containerRun}
c.setRuns(set)
for _, run := range set {
c.n += int32(run.last-run.start) + 1
c.n += int32(run.Last-run.Start) + 1
}
return c
}
// NewContainerRunCopy creates a new run container using a provided (possibly nil)
// slice of intervals. It copies the provided slice to new storage.
func NewContainerRunCopy(set []interval16) *Container {
func NewContainerRunCopy(set []Interval16) *Container {
c := &Container{typeID: containerRun}
c.setRunsMaybeCopy(set, true)
for _, run := range set {
c.n += int32(run.last-run.start) + 1
c.n += int32(run.Last-run.Start) + 1
}
return c
}
// NewContainerRunN creates a new run array using a provided (possibly nil)
// slice of intervals. It overrides n using the provided value.
func NewContainerRunN(set []interval16, n int32) *Container {
func NewContainerRunN(set []Interval16, n int32) *Container {
c := &Container{typeID: containerRun, n: n}
c.setRuns(set)
return c
@ -426,27 +426,33 @@ var fillerBitmap = func() (a [1024]uint64) {
return a
}()
func splatRun(into *[1024]uint64, from interval16) {
func splatRun(into *[1024]uint64, from Interval16) {
// TODO this can be ~64x faster for long runs by setting maxBitmap instead of single bits
// note v must be int or will overflow
// for v := int(from.Start); v <= int(from.Last); v++ {
// into[v/64] |= (uint64(1) << uint(v%64))
// }
// Handle the case where the start and end fall within the same word.
if from.start/64 == from.last/64 {
highMask := ^uint64(0) >> (63 - (from.last % 64))
lowMask := ^uint64(0) << (from.start % 64)
into[from.start/64] |= highMask & lowMask
if from.Start/64 == from.Last/64 {
highMask := ^uint64(0) >> (63 - (from.Last % 64))
lowMask := ^uint64(0) << (from.Start % 64)
into[from.Start/64] |= highMask & lowMask
return
}
// Calculate preliminary bulk fill bounds.
fillStart, fillEnd := from.start/64, from.last/64
fillStart, fillEnd := from.Start/64, from.Last/64
// Handle run start.
if from.start%64 != 0 {
into[from.start/64] |= ^uint64(0) << (from.start % 64)
if from.Start%64 != 0 {
into[from.Start/64] |= ^uint64(0) << (from.Start % 64)
fillStart++
}
// Handle run end.
if from.last%64 != 63 {
into[from.last/64] |= ^uint64(0) >> (63 - (from.last % 64))
if from.Last%64 != 63 {
into[from.Last/64] |= ^uint64(0) >> (63 - (from.Last % 64))
fillEnd--
}
@ -480,7 +486,7 @@ func (c *Container) setBitmap(bitmap []uint64) {
}
// runs yields the data viewed as a slice of intervals.
func (c *Container) runs() []interval16 {
func (c *Container) runs() []Interval16 {
if c == nil {
panic("attempt to read nil container's runs")
}
@ -489,17 +495,17 @@ func (c *Container) runs() []interval16 {
panic("attempt to read non-run's runs")
}
}
return (*[1 << 15]interval16)(unsafe.Pointer(c.pointer))[:c.len:c.cap]
return (*[1 << 15]Interval16)(unsafe.Pointer(c.pointer))[:c.len:c.cap]
}
// setRuns stores a set of intervals as data. c must not be frozen.
func (c *Container) setRuns(runs []interval16) {
func (c *Container) setRuns(runs []Interval16) {
c.setRunsMaybeCopy(runs, false)
}
// setRunsMaybeCopy stores a set of intervals as data. c must not be frozen.
// If doCopy is set, the values will be copied to different storage.
func (c *Container) setRunsMaybeCopy(runs []interval16, doCopy bool) {
func (c *Container) setRunsMaybeCopy(runs []Interval16, doCopy bool) {
if roaringParanoia {
if c == nil || c.frozen() {
panic("setRuns on nil or frozen container")
@ -514,24 +520,24 @@ func (c *Container) setRunsMaybeCopy(runs []interval16, doCopy bool) {
c.flags &^= flagPristine
// array we can fit in data store:
if len(runs) <= stashedRunSize {
newRuns := (*[stashedRunSize]interval16)(unsafe.Pointer(&c.data))[:len(runs)]
newRuns := (*[stashedRunSize]Interval16)(unsafe.Pointer(&c.data))[:len(runs)]
copy(newRuns, runs)
c.pointer, c.len, c.cap = &c.data[0], int32(len(newRuns)), int32(cap(newRuns))
c.flags &^= flagMapped // this is no longer using a hypothetical mmapped input array
return
}
if &runs[0].start == c.pointer && !doCopy {
if &runs[0].Start == c.pointer && !doCopy {
// nothing to do but update length
c.len = int32(len(runs))
return
}
if doCopy {
runs = append([]interval16(nil), runs...)
runs = append([]Interval16(nil), runs...)
}
if cap(runs) > 1<<15 {
runs = runs[: len(runs) : 1<<15]
}
c.pointer, c.len, c.cap = &runs[0].start, int32(len(runs)), int32(cap(runs))
c.pointer, c.len, c.cap = &runs[0].Start, int32(len(runs)), int32(cap(runs))
}
// UpdateOrMake updates the container, yielding a new container if necessary.

24
roaring/containers_test.go
View file

@ -187,18 +187,18 @@ func TestSliceContainers(t *testing.T) {
})
}
func genRun(r *rand.Rand) interval16 {
func genRun(r *rand.Rand) Interval16 {
gen:
dat := r.Uint32()
start, end := uint16(dat>>16), uint16(dat)
if start > end {
goto gen
}
return interval16{start, end}
return Interval16{start, end}
}
func splatRunNaive(into []uint64, from interval16) {
for v := int(from.start); v <= int(from.last); v++ {
func splatRunNaive(into []uint64, from Interval16) {
for v := int(from.Start); v <= int(from.Last); v++ {
into[v/64] |= (uint64(1) << uint(v%64))
}
}
@ -212,21 +212,21 @@ func TestSplat(t *testing.T) {
splatRunNaive(a[:], run)
splatRun(&b, run)
if a != b {
t.Errorf("incorrect splat of run [%d, %d]", run.start, run.last)
t.Errorf("incorrect splat of run [%d, %d]", run.Start, run.Last)
}
}
}
func benchSplat(b *testing.B, run interval16) {
func benchSplat(b *testing.B, run Interval16) {
var buf [1024]uint64
for i := 0; i < b.N; i++ {
splatRun(&buf, run)
}
}
func BenchmarkSplatSingle(b *testing.B) { benchSplat(b, interval16{42, 42}) }
func BenchmarkSplatPartword(b *testing.B) { benchSplat(b, interval16{16, 31}) }
func BenchmarkSplatWord(b *testing.B) { benchSplat(b, interval16{16, 31}) }
func BenchmarkSplatEdges(b *testing.B) { benchSplat(b, interval16{15, 16}) }
func BenchmarkSplatMedium(b *testing.B) { benchSplat(b, interval16{13, 65}) }
func BenchmarkSplatAll(b *testing.B) { benchSplat(b, interval16{0, ^uint16(0)}) }
func BenchmarkSplatSingle(b *testing.B) { benchSplat(b, Interval16{42, 42}) }
func BenchmarkSplatPartword(b *testing.B) { benchSplat(b, Interval16{16, 31}) }
func BenchmarkSplatWord(b *testing.B) { benchSplat(b, Interval16{16, 31}) }
func BenchmarkSplatEdges(b *testing.B) { benchSplat(b, Interval16{15, 16}) }
func BenchmarkSplatMedium(b *testing.B) { benchSplat(b, Interval16{13, 65}) }
func BenchmarkSplatAll(b *testing.B) { benchSplat(b, Interval16{0, ^uint16(0)}) }

651
roaring/roaring.go
View file

File diff suppressed because it is too large Load diff

62
roaring/roaring_helpers_test.go
View file

@ -175,65 +175,65 @@ func bitmapEvenBitsSet() []uint64 {
}
////////////////// run
func runEmpty() []interval16 {
return make([]interval16, 0)
func runEmpty() []Interval16 {
return make([]Interval16, 0)
}
func runFull() []interval16 {
run := make([]interval16, 0)
run = append(run, interval16{start: 0, last: 65535})
func runFull() []Interval16 {
run := make([]Interval16, 0)
run = append(run, Interval16{Start: 0, Last: 65535})
return run
}
func runFirstBitSet() []interval16 {
run := make([]interval16, 0)
run = append(run, interval16{start: 0, last: 0})
func runFirstBitSet() []Interval16 {
run := make([]Interval16, 0)
run = append(run, Interval16{Start: 0, Last: 0})
return run
}
func runLastBitSet() []interval16 {
run := make([]interval16, 0)
run = append(run, interval16{start: 65535, last: 65535})
func runLastBitSet() []Interval16 {
run := make([]Interval16, 0)
run = append(run, Interval16{Start: 65535, Last: 65535})
return run
}
func runFirstBitUnset() []interval16 {
run := make([]interval16, 0)
run = append(run, interval16{start: 1, last: 65535})
func runFirstBitUnset() []Interval16 {
run := make([]Interval16, 0)
run = append(run, Interval16{Start: 1, Last: 65535})
return run
}
func runLastBitUnset() []interval16 {
run := make([]interval16, 0)
run = append(run, interval16{start: 0, last: 65534})
func runLastBitUnset() []Interval16 {
run := make([]Interval16, 0)
run = append(run, Interval16{Start: 0, Last: 65534})
return run
}
func runInnerBitsSet() []interval16 {
run := make([]interval16, 0)
run = append(run, interval16{start: 1, last: 65534})
func runInnerBitsSet() []Interval16 {
run := make([]Interval16, 0)
run = append(run, Interval16{Start: 1, Last: 65534})
return run
}
func runOuterBitsSet() []interval16 {
run := make([]interval16, 0)
run = append(run, interval16{start: 0, last: 0})
run = append(run, interval16{start: 65535, last: 65535})
func runOuterBitsSet() []Interval16 {
run := make([]Interval16, 0)
run = append(run, Interval16{Start: 0, Last: 0})
run = append(run, Interval16{Start: 65535, Last: 65535})
return run
}
func runOddBitsSet() []interval16 {
run := make([]interval16, containerWidth/2)
func runOddBitsSet() []Interval16 {
run := make([]Interval16, containerWidth/2)
for i := 0; i < int(containerWidth/2); i++ {
run[i] = interval16{start: uint16(2*i + 1), last: uint16(2*i + 1)}
run[i] = Interval16{Start: uint16(2*i + 1), Last: uint16(2*i + 1)}
}
return run
}
func runEvenBitsSet() []interval16 {
run := make([]interval16, containerWidth/2)
func runEvenBitsSet() []Interval16 {
run := make([]Interval16, containerWidth/2)
for i := 0; i < int(containerWidth/2); i++ {
run[i] = interval16{start: uint16(2 * i), last: uint16(2 * i)}
run[i] = Interval16{Start: uint16(2 * i), Last: uint16(2 * i)}
}
return run
}
@ -258,7 +258,7 @@ func doContainer(typ byte, data interface{}) *Container {
c := NewContainerBitmap(-1, data.([]uint64))
return c
case containerRun:
return NewContainerRun(data.([]interval16))
return NewContainerRun(data.([]Interval16))
}
return nil
}

771
roaring/roaring_internal_test.go
View file

File diff suppressed because it is too large Load diff

6
roaring/roaring_test.go
View file

@ -315,8 +315,9 @@ func TestBitmap_SliceRange(t *testing.T) {
// Ensure a bitmap can loop over a set of values.
func TestBitmap_ForEach(t *testing.T) {
var a []uint64
roaring.NewFileBitmap(1, 2, 3).ForEach(func(v uint64) {
_ = roaring.NewFileBitmap(1, 2, 3).ForEach(func(v uint64) error {
a = append(a, v)
return nil
})
if !reflect.DeepEqual(a, []uint64{1, 2, 3}) {
t.Fatalf("unexpected values: %+v", a)
@ -326,8 +327,9 @@ func TestBitmap_ForEach(t *testing.T) {
// Ensure a bitmap can loop over a set of values in a range.
func TestBitmap_ForEachRange(t *testing.T) {
var a []uint64
roaring.NewFileBitmap(1, 2, 3, 4).ForEachRange(2, 4, func(v uint64) {
_ = roaring.NewFileBitmap(1, 2, 3, 4).ForEachRange(2, 4, func(v uint64) error {
a = append(a, v)
return nil
})
if !reflect.DeepEqual(a, []uint64{2, 3}) {
t.Fatalf("unexpected values: %+v", a)

4
roaring/unmarshal_binary.go
View file

@ -52,7 +52,7 @@ func (b *Bitmap) UnmarshalBinary(data []byte) (err error) {
case containerArray:
newC = NewContainerArray((*[4096]uint16)(unsafe.Pointer(itrPointer))[:itrLen:itrLen])
case containerRun:
newC = NewContainerRunN((*[2048]interval16)(unsafe.Pointer(itrPointer))[:itrLen:itrLen], int32(itrN))
newC = NewContainerRunN((*[2048]Interval16)(unsafe.Pointer(itrPointer))[:itrLen:itrLen], int32(itrN))
case containerBitmap:
newC = NewContainerBitmapN((*[1024]uint64)(unsafe.Pointer(itrPointer))[:1024:itrLen], int32(itrN))
default:
@ -144,7 +144,7 @@ func InspectBinary(data []byte, mapped bool, info *BitmapInfo) (b *Bitmap, mappe
case containerArray:
newC = NewContainerArray((*[4096]uint16)(unsafe.Pointer(itrPointer))[:itrLen:itrLen])
case containerRun:
newC = NewContainerRunN((*[2048]interval16)(unsafe.Pointer(itrPointer))[:itrLen:itrLen], int32(itrN))
newC = NewContainerRunN((*[2048]Interval16)(unsafe.Pointer(itrPointer))[:itrLen:itrLen], int32(itrN))
case containerBitmap:
newC = NewContainerBitmapN((*[1024]uint64)(unsafe.Pointer(itrPointer))[:1024:itrLen], int32(itrN))
default:

7
server/grpc.go
View file

@ -272,6 +272,9 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errToStatusError(err)
}
// Obtain transaction.
tx := pilosa.NewMultiTxWithIndex(true, index)
var fields []*pilosa.Field
for _, field := range index.Fields() {
// exclude internal fields (starting with "_")
@ -440,7 +443,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
}
}
} else {
value, exists, err = field.StringValue(col)
value, exists, err = field.StringValue(tx, col)
if err != nil {
return errors.Wrap(err, "getting string field value for column")
}
@ -689,7 +692,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
}
}
} else {
value, exists, err = field.StringValue(id)
value, exists, err = field.StringValue(tx, id)
if err != nil {
return errors.Wrap(err, "getting string field value for column")
}

14
server/handler_test.go
View file

@ -173,22 +173,32 @@ func TestHandler_Endpoints(t *testing.T) {
}
})
tx, err := holder.Begin(true)
if err != nil {
t.Fatal(err)
}
defer func() { _ = tx.Rollback() }()
i0 := hldr.MustCreateIndexIfNotExists("i0", pilosa.IndexOptions{})
i1 := hldr.MustCreateIndexIfNotExists("i1", pilosa.IndexOptions{})
if f, err := i0.CreateFieldIfNotExists("f1", pilosa.OptFieldTypeDefault()); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(0, 0, nil); err != nil {
} else if _, err := f.SetBit(tx, 0, 0, nil); err != nil {
t.Fatal(err)
}
if f, err := i1.CreateFieldIfNotExists("f0", pilosa.OptFieldTypeDefault()); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(0, 0, nil); err != nil {
} else if _, err := f.SetBit(tx, 0, 0, nil); err != nil {
t.Fatal(err)
}
if _, err := i0.CreateFieldIfNotExists("f0", pilosa.OptFieldTypeDefault()); err != nil {
t.Fatal(err)
}
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
t.Run("Schema", func(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/schema", nil))

32
test/holder.go
View file

@ -86,7 +86,9 @@ func (h *Holder) Row(index, field string, rowID uint64) *pilosa.Row {
if err != nil {
panic(err)
}
row, err := f.Row(rowID)
tx := &pilosa.RoaringTx{Index: idx.Index}
row, err := f.Row(tx, rowID)
if err != nil {
panic(err)
}
@ -100,7 +102,9 @@ func (h *Holder) ReadRow(index, field string, rowID uint64) *pilosa.Row {
if f == nil {
panic(pilosa.ErrFieldNotFound)
}
row, err := f.Row(rowID)
tx := &pilosa.RoaringTx{Field: f}
row, err := f.Row(tx, rowID)
if err != nil {
panic(err)
}
@ -122,7 +126,9 @@ func (h *Holder) RowTime(index, field string, rowID uint64, t time.Time, quantum
if err != nil {
panic(err)
}
row, err := f.RowTime(rowID, t, quantum)
tx := &pilosa.RoaringTx{Index: idx.Index}
row, err := f.RowTime(tx, rowID, t, quantum)
if err != nil {
panic(err)
}
@ -141,7 +147,9 @@ func (h *Holder) SetBitTime(index, field string, rowID, columnID uint64, t *time
if err != nil {
panic(err)
}
_, err = f.SetBit(rowID, columnID, t)
tx := &pilosa.RoaringTx{Index: idx.Index}
_, err = f.SetBit(tx, rowID, columnID, t)
if err != nil {
panic(err)
}
@ -154,7 +162,9 @@ func (h *Holder) ClearBit(index, field string, rowID, columnID uint64) {
if err != nil {
panic(err)
}
_, err = f.ClearBit(rowID, columnID)
tx := &pilosa.RoaringTx{Index: idx.Index}
_, err = f.ClearBit(tx, rowID, columnID)
if err != nil {
panic(err)
}
@ -175,7 +185,9 @@ func (h *Holder) SetValue(index, field string, columnID uint64, value int64) {
if err != nil {
panic(err)
}
_, err = f.SetValue(columnID, value)
tx := &pilosa.RoaringTx{Index: idx.Index}
_, err = f.SetValue(tx, columnID, value)
if err != nil {
panic(err)
}
@ -188,7 +200,9 @@ func (h *Holder) Value(index, field string, columnID uint64) (int64, bool) {
if err != nil {
panic(err)
}
val, exists, err := f.Value(columnID)
tx := &pilosa.RoaringTx{Index: idx.Index}
val, exists, err := f.Value(tx, columnID)
if err != nil {
panic(err)
}
@ -203,7 +217,9 @@ func (h *Holder) Range(index, field string, op pql.Token, predicate int64) *pilo
if err != nil {
panic(err)
}
row, err := f.Range(field, op, predicate)
tx := &pilosa.RoaringTx{Index: idx.Index}
row, err := f.Range(tx, field, op, predicate)
if err != nil {
panic(err)
}

433
tx.go Normal file
View file

@ -0,0 +1,433 @@
// Copyright 2017 Pilosa 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
import (
"fmt"
"sync"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pkg/errors"
)
type Tx interface {
Rollback() error
Commit() error
RoaringBitmap(index, field, view string, shard uint64) (*roaring.Bitmap, error)
Container(index, field, view string, shard uint64, key uint64) (*roaring.Container, error)
PutContainer(index, field, view string, shard uint64, key uint64, c *roaring.Container) error
RemoveContainer(index, field, view string, shard uint64, key uint64) error
Add(index, field, view string, shard uint64, a ...uint64) (changed bool, err error)
Remove(index, field, view string, shard uint64, a ...uint64) (changed bool, err error)
Contains(index, field, view string, shard uint64, v uint64) (exists bool, err error)
ContainerIterator(index, field, view string, shard uint64, key uint64) (citer roaring.ContainerIterator, found bool, err error)
ForEach(index, field, view string, shard uint64, fn func(i uint64) error) error
ForEachRange(index, field, view string, shard uint64, start, end uint64, fn func(uint64) error) error
Count(index, field, view string, shard uint64) (uint64, error)
Max(index, field, view string, shard uint64) (uint64, error)
Min(index, field, view string, shard uint64) (uint64, bool, error)
UnionInPlace(index, field, view string, shard uint64, others ...*roaring.Bitmap) error
CountRange(index, field, view string, shard uint64, start, end uint64) (uint64, error)
OffsetRange(index, field, view string, shard uint64, offset, start, end uint64) (*roaring.Bitmap, error)
}
// MultiTx implements the transaction interface to combine multiple transactions.
type MultiTx struct {
mu sync.Mutex
writable bool
holder *Holder
index *Index
txs map[multiTxKey]Tx
}
// NewMultiTx returns a new instance of MultiTx for a Holder.
func NewMultiTx(writable bool, holder *Holder) *MultiTx {
return &MultiTx{
writable: writable,
holder: holder,
txs: make(map[multiTxKey]Tx),
}
}
// NewMultiTxWithIndex returns a new instance of MultiTx for a single index.
func NewMultiTxWithIndex(writable bool, index *Index) *MultiTx {
return &MultiTx{
writable: writable,
index: index,
txs: make(map[multiTxKey]Tx),
}
}
var _ Tx = (*MultiTx)(nil)
// Rollback rolls back all underlying transactions.
func (mtx *MultiTx) Rollback() (err error) {
for _, tx := range mtx.txs {
if e := tx.Rollback(); e != nil && err == nil {
err = e
}
}
return err
}
// Commit commits all underlying transactions.
func (mtx *MultiTx) Commit() (err error) {
for _, tx := range mtx.txs {
if e := tx.Commit(); e != nil && err == nil {
err = e
}
}
return err
}
func (mtx *MultiTx) RoaringBitmap(index, field, view string, shard uint64) (*roaring.Bitmap, error) {
tx, err := mtx.tx(index, shard)
if err != nil {
return nil, err
}
return tx.RoaringBitmap(index, field, view, shard)
}
func (mtx *MultiTx) Container(index, field, view string, shard uint64, key uint64) (*roaring.Container, error) {
tx, err := mtx.tx(index, shard)
if err != nil {
return nil, err
}
return tx.Container(index, field, view, shard, key)
}
func (mtx *MultiTx) PutContainer(index, field, view string, shard uint64, key uint64, c *roaring.Container) error {
tx, err := mtx.tx(index, shard)
if err != nil {
return err
}
return tx.PutContainer(index, field, view, shard, key, c)
}
func (mtx *MultiTx) RemoveContainer(index, field, view string, shard uint64, key uint64) error {
tx, err := mtx.tx(index, shard)
if err != nil {
return err
}
return tx.RemoveContainer(index, field, view, shard, key)
}
func (mtx *MultiTx) Add(index, field, view string, shard uint64, a ...uint64) (changed bool, err error) {
tx, err := mtx.tx(index, shard)
if err != nil {
return false, err
}
return tx.Add(index, field, view, shard, a...)
}
func (mtx *MultiTx) Remove(index, field, view string, shard uint64, a ...uint64) (changed bool, err error) {
tx, err := mtx.tx(index, shard)
if err != nil {
return false, err
}
return tx.Remove(index, field, view, shard, a...)
}
func (mtx *MultiTx) Contains(index, field, view string, shard uint64, v uint64) (exists bool, err error) {
tx, err := mtx.tx(index, shard)
if err != nil {
return false, err
}
return tx.Contains(index, field, view, shard, v)
}
func (mtx *MultiTx) ContainerIterator(index, field, view string, shard uint64, key uint64) (citer roaring.ContainerIterator, found bool, err error) {
tx, err := mtx.tx(index, shard)
if err != nil {
return nil, false, err
}
return tx.ContainerIterator(index, field, view, shard, key)
}
func (mtx *MultiTx) ForEach(index, field, view string, shard uint64, fn func(i uint64) error) error {
tx, err := mtx.tx(index, shard)
if err != nil {
return err
}
return tx.ForEach(index, field, view, shard, fn)
}
func (mtx *MultiTx) ForEachRange(index, field, view string, shard uint64, start, end uint64, fn func(uint64) error) error {
tx, err := mtx.tx(index, shard)
if err != nil {
return err
}
return tx.ForEachRange(index, field, view, shard, start, end, fn)
}
func (mtx *MultiTx) Count(index, field, view string, shard uint64) (uint64, error) {
tx, err := mtx.tx(index, shard)
if err != nil {
return 0, err
}
return tx.Count(index, field, view, shard)
}
func (mtx *MultiTx) Max(index, field, view string, shard uint64) (uint64, error) {
tx, err := mtx.tx(index, shard)
if err != nil {
return 0, err
}
return tx.Max(index, field, view, shard)
}
func (mtx *MultiTx) Min(index, field, view string, shard uint64) (uint64, bool, error) {
tx, err := mtx.tx(index, shard)
if err != nil {
return 0, false, err
}
return tx.Min(index, field, view, shard)
}
func (mtx *MultiTx) UnionInPlace(index, field, view string, shard uint64, others ...*roaring.Bitmap) error {
tx, err := mtx.tx(index, shard)
if err != nil {
return err
}
return tx.UnionInPlace(index, field, view, shard, others...)
}
func (mtx *MultiTx) CountRange(index, field, view string, shard uint64, start, end uint64) (uint64, error) {
tx, err := mtx.tx(index, shard)
if err != nil {
return 0, err
}
return tx.CountRange(index, field, view, shard, start, end)
}
func (mtx *MultiTx) OffsetRange(index, field, view string, shard uint64, offset, start, end uint64) (*roaring.Bitmap, error) {
tx, err := mtx.tx(index, shard)
if err != nil {
return nil, err
}
return tx.OffsetRange(index, field, view, shard, offset, start, end)
}
// tx returns a transaction by index/shard. Reuses transaction if already open.
// Otherwise begins a new transaction.
func (mtx *MultiTx) tx(index string, shard uint64) (_ Tx, err error) {
mtx.mu.Lock()
defer mtx.mu.Unlock()
// Lookup transaction from cache.
tx := mtx.txs[multiTxKey{index, shard}]
if tx != nil {
return tx, nil
}
// If transaction doesn't exist, lookup the index.
idx := mtx.index
if mtx.holder != nil {
if idx = mtx.holder.Index(index); idx == nil {
return nil, ErrIndexNotFound
}
}
// Begin tranaction & cache it.
if tx, err = idx.Begin(mtx.writable, shard); err != nil {
return nil, err
}
mtx.txs[multiTxKey{index, shard}] = tx
return tx, nil
}
type multiTxKey struct {
index string
shard uint64
}
// RoaringTx represents a fake transaction object for Roaring storage.
type RoaringTx struct {
Index *Index
Field *Field
fragment *fragment
}
// Rollback is a no-op as Roaring does not support transactions.
func (tx *RoaringTx) Rollback() error {
return nil
}
// Commit is a no-op as Roaring does not support transactions.
func (tx *RoaringTx) Commit() error {
return nil
}
func (tx *RoaringTx) RoaringBitmap(index, field, view string, shard uint64) (*roaring.Bitmap, error) {
return tx.bitmap(field, view, shard)
}
func (tx *RoaringTx) Container(index, field, view string, shard uint64, key uint64) (*roaring.Container, error) {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return nil, err
}
return b.Containers.Get(key), nil
}
func (tx *RoaringTx) PutContainer(index, field, view string, shard uint64, key uint64, c *roaring.Container) error {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return err
}
b.Containers.Put(key, c)
return nil
}
func (tx *RoaringTx) RemoveContainer(index, field, view string, shard uint64, key uint64) error {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return err
}
b.Containers.Remove(key)
return nil
}
func (tx *RoaringTx) Add(index, field, view string, shard uint64, a ...uint64) (changed bool, err error) {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return false, err
}
return b.Add(a...)
}
func (tx *RoaringTx) Remove(index, field, view string, shard uint64, a ...uint64) (changed bool, err error) {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return false, err
}
return b.Remove(a...)
}
func (tx *RoaringTx) Contains(index, field, view string, shard uint64, v uint64) (exists bool, err error) {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return false, err
}
return b.Contains(v), nil
}
func (tx *RoaringTx) ContainerIterator(index, field, view string, shard uint64, key uint64) (citer roaring.ContainerIterator, found bool, err error) {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return nil, false, err
}
citer, found = b.Containers.Iterator(key)
return citer, found, nil
}
func (tx *RoaringTx) ForEach(index, field, view string, shard uint64, fn func(i uint64) error) error {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return err
}
return b.ForEach(fn)
}
func (tx *RoaringTx) ForEachRange(index, field, view string, shard uint64, start, end uint64, fn func(uint64) error) error {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return err
}
return b.ForEachRange(start, end, fn)
}
func (tx *RoaringTx) Count(index, field, view string, shard uint64) (uint64, error) {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return 0, err
}
return b.Count(), nil
}
func (tx *RoaringTx) Max(index, field, view string, shard uint64) (uint64, error) {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return 0, err
}
return b.Max(), nil
}
func (tx *RoaringTx) Min(index, field, view string, shard uint64) (uint64, bool, error) {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return 0, false, err
}
v, ok := b.Min()
return v, ok, nil
}
func (tx *RoaringTx) UnionInPlace(index, field, view string, shard uint64, others ...*roaring.Bitmap) error {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return err
}
b.UnionInPlace(others...)
return nil
}
func (tx *RoaringTx) CountRange(index, field, view string, shard uint64, start, end uint64) (uint64, error) {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return 0, err
}
return b.CountRange(start, end), nil
}
func (tx *RoaringTx) OffsetRange(index, field, view string, shard uint64, offset, start, end uint64) (*roaring.Bitmap, error) {
b, err := tx.bitmap(field, view, shard)
if err != nil {
return nil, err
}
return b.OffsetRange(offset, start, end), nil
}
func (tx *RoaringTx) bitmap(field, view string, shard uint64) (*roaring.Bitmap, error) {
// If a fragment is attached, always use it.
if tx.fragment != nil {
return tx.fragment.storage, nil
}
// If a field is attached, start from there.
// Otherwise look up the field from the index.
f := tx.Field
if f == nil {
if f = tx.Index.Field(field); f == nil {
return nil, ErrFieldNotFound
}
}
v := f.view(view)
if v == nil {
return nil, errors.Errorf("view not found: %q", view)
}
frag := v.Fragment(shard)
if frag == nil {
panic(fmt.Sprintf("fragment not found: %q / %q / %d", field, view, shard))
}
return frag.storage, nil
}

17
utils_internal_test.go
View file

@ -133,8 +133,21 @@ func (t *ClusterCluster) SetBit(index, field string, rowID, colID uint64, x *tim
if f == nil {
return fmt.Errorf("index/field does not exist: %s/%s", index, field)
}
_, err := f.SetBit(rowID, colID, x)
if err != nil {
if err := func() error {
tx, err := c.holder.Begin(true)
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
if _, err := f.SetBit(tx, rowID, colID, x); err != nil {
return err
} else if err := tx.Commit(); err != nil {
return err
}
return nil
}(); err != nil {
return err
}
}

34
view.go
View file

@ -402,74 +402,76 @@ func (v *view) deleteFragment(shard uint64) error {
}
// row returns a row for a shard of the view.
func (v *view) row(rowID uint64) *Row {
func (v *view) row(tx Tx, rowID uint64) (*Row, error) {
row := NewRow()
for _, frag := range v.allFragments() {
fr := frag.row(rowID)
if fr == nil {
fr, err := frag.row(tx, rowID)
if err != nil {
return nil, err
} else if fr == nil {
continue
}
row.Merge(fr)
}
return row
return row, nil
}
// setBit sets a bit within the view.
func (v *view) setBit(rowID, columnID uint64) (changed bool, err error) {
func (v *view) setBit(tx Tx, rowID, columnID uint64) (changed bool, err error) {
shard := columnID / ShardWidth
frag, err := v.CreateFragmentIfNotExists(shard)
if err != nil {
return changed, err
}
return frag.setBit(rowID, columnID)
return frag.setBit(tx, rowID, columnID)
}
// clearBit clears a bit within the view.
func (v *view) clearBit(rowID, columnID uint64) (changed bool, err error) {
func (v *view) clearBit(tx Tx, rowID, columnID uint64) (changed bool, err error) {
shard := columnID / ShardWidth
frag := v.Fragment(shard)
if frag == nil {
return false, nil
}
return frag.clearBit(rowID, columnID)
return frag.clearBit(tx, rowID, columnID)
}
// value uses a column of bits to read a multi-bit value.
func (v *view) value(columnID uint64, bitDepth uint) (value int64, exists bool, err error) {
func (v *view) value(tx Tx, columnID uint64, bitDepth uint) (value int64, exists bool, err error) {
shard := columnID / ShardWidth
frag, err := v.CreateFragmentIfNotExists(shard)
if err != nil {
return value, exists, err
}
return frag.value(columnID, bitDepth)
return frag.value(tx, columnID, bitDepth)
}
// setValue uses a column of bits to set a multi-bit value.
func (v *view) setValue(columnID uint64, bitDepth uint, value int64) (changed bool, err error) {
func (v *view) setValue(tx Tx, columnID uint64, bitDepth uint, value int64) (changed bool, err error) {
shard := columnID / ShardWidth
frag, err := v.CreateFragmentIfNotExists(shard)
if err != nil {
return changed, err
}
return frag.setValue(columnID, bitDepth, value)
return frag.setValue(tx, columnID, bitDepth, value)
}
// clearValue removes a specific value assigned to columnID
func (v *view) clearValue(columnID uint64, bitDepth uint, value int64) (changed bool, err error) {
func (v *view) clearValue(tx Tx, columnID uint64, bitDepth uint, value int64) (changed bool, err error) {
shard := columnID / ShardWidth
frag := v.Fragment(shard)
if frag == nil {
return false, nil
}
return frag.clearValue(columnID, bitDepth, value)
return frag.clearValue(tx, columnID, bitDepth, value)
}
// rangeOp returns rows with a field value encoding matching the predicate.
func (v *view) rangeOp(op pql.Token, bitDepth uint, predicate int64) (*Row, error) {
func (v *view) rangeOp(tx Tx, op pql.Token, bitDepth uint, predicate int64) (*Row, error) {
r := NewRow()
for _, frag := range v.allFragments() {
other, err := frag.rangeOp(op, bitDepth, predicate)
other, err := frag.rangeOp(tx, op, bitDepth, predicate)
if err != nil {
return nil, err
}