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

significant refactor of test setup and teardown

Browse source 
We centralize the creation paths for test indexes, fields,
etcetera so they all have a common path, all using standard
test holders. There's still two versions, one for test.* functions
and one for internal. They do share a TestHolderConfig though.

Large hunks of the related APIs are simplified/streamlined.
* Fragments are always created with a Field and don't need
  a workaround in case they don't have it.
* Creation of test fragments, etc., use optional FieldOptions
  but don't specify names because they're all using new holders
  for each thing created anyway. This dramatically reduces
  the complexity of the calls.
* test fragments are created inside test views which are created
  inside test fields, etcetera, so everything is using the same
  logic; test views aren't bypassing the other layers, they're
  creating themselves normally within a field.
* Quite a few things now use the standard runtime/production
  logic instead of being custom workarounds; for instance, instead
  of `mustOpenMutexFragment` creating a fragment and then creating
  a mutex vector for it, we just create a mutex-typed field and
  have the normal runtime code do this.
* Similarly, we now use the same field creation logic that production
  does, instead of having our own test-only thing that validates
  field names directly, so our test that we're validating field names
  is actually testing the runtime code. Yay.
* fragSpec goes away. it was a replacement for fragProxy which existed
  to solve memory allocation problems but replaced them with interface
  overhead problems. Now we just have pointers to things and maintain
  valid data structures.
* Many panics are now Fatal or Fatalf calls.
* Some specific bugs fixed, like a cluster which was requested and
  then had its first node directly overwritten, which isn't valid with
  shared clusters.
* Drop the temp-dir test flag and TempDir variable, we can just use
  $TMPDIR.
* Drop a benchmark of "write file to disk" that was purely a benchmark
  of file write speed, not a benchmark of rendering the data that needs
  to be written.
* Drop the unused "flags" parameter to fragment creation, which was
  only used back when we changed the BSI format.
* Use holder.Txf() rather than index.Txf(). The TxFactory has to be
  holder-level anyway, referring to it via the index is misleading.
* Test holders automatically close themselves and delete themselves,
  we remove various other things that thought they were responsible
  for deleting themselves.
This commit is contained in:
Seebs 2022-09-21 14:09:21 -05:00 • committed by Christopher Lowenthal
parent aaf493bc1b
commit 9ca78d4ebd
28 changed files with 460 additions and 754 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
api.go
View file

@ -2643,7 +2643,7 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64
idx := api.holder.Index(indexName)
//need to get a dbShard
dbs, err := idx.Txf().dbPerShard.GetDBShard(indexName, shard, idx)
dbs, err := api.holder.Txf().dbPerShard.GetDBShard(indexName, shard, idx)
if err != nil {
return err
}

28
dbshard_internal_test.go
View file

@ -17,7 +17,7 @@ import (
// Shard per db evaluation
func TestShardPerDB_SetBit(t *testing.T) {
f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "")
f, idx, tx := mustOpenFragment(t)
_ = idx
defer f.Clean(t)
@ -72,9 +72,7 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) {
}
for _, src := range []string{"rbf"} {
cfg := mustHolderConfig()
cfg.StorageConfig.Backend = src
holder := NewHolder(tmpdir, cfg)
holder := newTestHolder(t)
index := "rick"
idx := makeSampleRoaringDir(t, tmpdir, index, src, 1, holder, v2s)
@ -110,12 +108,10 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) {
}
tx.Rollback()
}
holder.Close()
}
}
// data for Test_DBPerShard_GetShardsForIndex
//
var sampleRoaringDirList = map[string]string{"roaring": `
rick/fields/f/views/standard/fragments/215.cache
rick/fields/f/views/standard/fragments/221.cache
@ -252,28 +248,16 @@ func makeTxTestDBWithViewsShards(tb testing.TB, holder *Holder, idx *Index, exp
// test that rbf can give us a map[view]*shardSet
func Test_DBPerShard_GetFieldView2Shards_map_from_RBF(t *testing.T) {
tmpdir, err := testhook.TempDir(t, "Test_DBPerShard_GetFieldView2Shards_map_from_RBF")
PanicOn(err)
defer os.RemoveAll(tmpdir)
cfg := mustHolderConfig()
cfg.StorageConfig.Backend = "rbf"
cfg.StorageConfig.FsyncEnabled = false
holder := NewHolder(tmpdir, cfg)
defer holder.Close()
holder := newTestHolder(t)
index := "rick"
field := "f"
cim := &CreateIndexMessage{
Index: index,
CreatedAt: 0,
Meta: IndexOptions{},
idx, err := holder.CreateIndex(index, IndexOptions{})
if err != nil {
t.Fatalf("creating index: %v", err)
}
idx, err := holder.createIndex(cim, false)
PanicOn(err)
exp := NewFieldView2Shards()
stdShardSet := newShardSet()

23
executor.go
View file

@ -8706,14 +8706,19 @@ func (e *executor) executeDeleteRecords(ctx context.Context, qcx *Qcx, index str
}
func transactExistRow(ctx context.Context, idx *Index, shard uint64, frag *fragment, src *Row) (uint64, error) {
tx := idx.Txf().NewTx(Txo{Write: writable, Index: idx, Shard: shard})
holder := idx.Holder()
tx := holder.Txf().NewTx(Txo{Write: writable, Index: idx, Shard: shard})
rows, err := frag.rows(ctx, tx, 1)
if err != nil {
tx.Rollback()
return 0, err
}
rowID := uint64(len(rows) + 1)
// obtain a rowID which is higher than any currently present row ID.
rowID := uint64(1)
if len(rows) > 0 {
rowID = rows[len(rows)-1] + 1
}
_, err = frag.setRow(tx, src, rowID)
if err != nil {
tx.Rollback()
@ -8731,7 +8736,7 @@ func (e *executor) executeDeleteRecordFromShard(ctx context.Context, index strin
err = newNotFoundError(ErrIndexNotFound, index)
return
}
qcx := idx.Txf().NewQcx()
qcx := e.Holder.Txf().NewQcx()
// bmCall is a bitmap
row, err := e.executeBitmapCallShard(ctx, qcx, index, bmCall, shard)
qcx.Abort()
@ -8757,9 +8762,10 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id
var existenceFragment *fragment
var deletedRowID uint64
var commitor Commitor = &NopCommitor{}
var err error // store columns in exits field ToBeDelete row commited
var err error // store columns in exits field ToBeDelete row commited
holder := idx.Holder()
if normalFlow { // normalFlow is the standard path, "not normal" is recoverory
existenceFragment = idx.Holder().fragment(idx.Name(), existenceFieldName, viewStandard, shard)
existenceFragment = holder.fragment(idx.Name(), existenceFieldName, viewStandard, shard)
if existenceFragment == nil {
// no exists field
return false, errors.New("can't bulk delete without existence field")
@ -8774,7 +8780,7 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id
if err != nil {
return false, err
}
writeTx := idx.Txf().NewTx(Txo{Write: writable, Index: idx, Shard: shard})
writeTx := holder.Txf().NewTx(Txo{Write: writable, Index: idx, Shard: shard})
if err != nil {
return false, err
}
@ -8799,7 +8805,7 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id
err = er
}
if err != nil {
idx.Holder().Logger.Errorf("problems committing delete in rbf %v shard %v", err, shard)
holder.Logger.Errorf("problems committing delete in rbf %v shard %v", err, shard)
}
}()
@ -8839,7 +8845,8 @@ func DeleteRowsWithOutKeysFlow(ctx context.Context, columns *roaring.Bitmap, idx
var existenceFragment *fragment
var deletedRowID uint64
var commitor Commitor = &NopCommitor{}
writeTx := idx.Txf().NewTx(Txo{Write: writable, Index: idx, Shard: shard})
holder := idx.Holder()
writeTx := holder.Txf().NewTx(Txo{Write: writable, Index: idx, Shard: shard})
defer writeTx.Rollback()
defer func() {
// if there is an error in the key commit, then rollback the delete

29
executor_internal_test.go
View file

@ -12,9 +12,7 @@ import (
"testing"
"time"
"github.com/featurebasedb/featurebase/v3/pql"
"github.com/featurebasedb/featurebase/v3/testhook"
"github.com/stretchr/testify/assert"
"github.com/molecula/featurebase/v3/pql"
)
// AssertEqual checks a given RowIdentifiers against expected values.
@ -42,24 +40,19 @@ func (r *RowIdentifiers) AssertEqual(tb testing.TB, other *RowIdentifiers) {
}
func TestExecutor_TranslateRowsOnBool(t *testing.T) {
path, _ := testhook.TempDirInDir(t, *TempDir, "pilosa-executor-")
holder := NewHolder(path, mustHolderConfig())
defer holder.Close()
holder := newTestHolder(t)
e := &executor{
Holder: holder,
Cluster: NewTestCluster(t, 1),
}
if err := e.Holder.Open(); err != nil {
t.Fatalf("opening holder: %v", err)
}
idx, err := e.Holder.CreateIndex("i", IndexOptions{})
if err != nil {
t.Fatalf("creating index: %v", err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := holder.Txf().NewWritableQcx()
defer qcx.Abort()
fb, errb := idx.CreateField("b", OptFieldTypeBool())
@ -513,10 +506,10 @@ func TestExecutorSafeCopyDistinctTimestamp(t *testing.T) {
}
func TestGetScaledInt(t *testing.T) {
f := OpenField(t, OptFieldTypeTimestamp(time.Now(), "ms"))
_, _, f := newTestField(t, OptFieldTypeTimestamp(time.Now(), "ms"))
// check that fields with type timestamp return the int64 passed in to getScaledInt with nil err
v := time.Now().Unix()
res, err := getScaledInt(f.Field, v)
res, err := getScaledInt(f, v)
if err != nil {
t.Errorf("got error %v, expected nil", err)
}
@ -568,25 +561,19 @@ func TestDistinctTimestampUnion(t *testing.T) {
}
func TestExecutor_DeleteRows(t *testing.T) {
path, _ := testhook.TempDir(t, "pilosa-executor-")
holder := NewHolder(path, mustHolderConfig())
defer holder.Close()
if err := holder.Open(); err != nil {
t.Fatalf("opening holder: %v", err)
}
holder := newTestHolder(t)
idx, err := holder.CreateIndex("i", IndexOptions{TrackExistence: true})
if err != nil {
t.Fatalf("creating index: %v", err)
}
f, err := idx.CreateField("f", OptFieldTypeDefault())
f, err := idx.CreateField("f")
if err != nil {
t.Fatalf("creating field: %v", err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := holder.Txf().NewWritableQcx()
defer qcx.Abort()
if _, err = f.SetBit(qcx, 1, 1, nil); err != nil {
t.Fatalf("setting bit: %v", err)

68
executor_test.go
View file

@ -1447,7 +1447,7 @@ func TestExecutor_Execute_Set(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("f", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := idx.CreateField("f"); err != nil {
t.Fatal(err)
}
@ -1464,7 +1464,7 @@ func TestExecutor_Execute_Set(t *testing.T) {
t.Run("ErrInvalidRowValueType", func(t *testing.T) {
idx := hldr.MustCreateIndexIfNotExists(c.Idx("inokey"), pilosa.IndexOptions{})
if _, err := idx.CreateField("f", pilosa.OptFieldTypeDefault(), pilosa.OptFieldKeys()); err != nil {
if _, err := idx.CreateField("f", pilosa.OptFieldKeys()); err != nil {
t.Fatal(err)
}
if _, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: c.Idx("inokey"), Query: `Set(2, f=1.2)`}); err == nil || !strings.Contains(err.Error(), "invalid value") {
@ -1635,7 +1635,7 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
index := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{})
if _, err := index.CreateFieldIfNotExists("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
} else if _, err := index.CreateFieldIfNotExists("xxx", pilosa.OptFieldTypeDefault()); err != nil {
} else if _, err := index.CreateFieldIfNotExists("xxx"); err != nil {
t.Fatal(err)
}
@ -1647,8 +1647,7 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
}
// Obtain transaction.
idx := index.Index
qcx := idx.Txf().NewQcx()
qcx := hldr.Txf().NewQcx()
defer qcx.Abort()
f := hldr.Field(c.Idx(), "f")
@ -1707,7 +1706,7 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
index := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{})
if _, err := index.CreateFieldIfNotExists("f", pilosa.OptFieldTypeTimestamp(pilosa.DefaultEpoch, pilosa.TimeUnitSeconds)); err != nil {
t.Fatal(err)
} else if _, err := index.CreateFieldIfNotExists("xxx", pilosa.OptFieldTypeDefault()); err != nil {
} else if _, err := index.CreateFieldIfNotExists("xxx"); err != nil {
t.Fatal(err)
}
@ -1719,8 +1718,7 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
}
// Obtain transaction.
idx := index.Index
qcx := idx.Txf().NewQcx()
qcx := hldr.Txf().NewQcx()
defer qcx.Abort()
f := hldr.Field(c.Idx(), "f")
@ -1847,9 +1845,9 @@ func TestExecutor_Execute_TopN(t *testing.T) {
// Set columns for rows 0, 10, & 20 across two shards.
if idx, err := hldr.CreateIndex(c.Idx(), pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
} else if _, err := idx.CreateField("f", pilosa.OptFieldTypeDefault()); err != nil {
} else if _, err := idx.CreateField("f"); err != nil {
t.Fatal(err)
} else if _, err := idx.CreateField("other", pilosa.OptFieldTypeDefault()); err != nil {
} else if _, err := idx.CreateField("other"); err != nil {
t.Fatal(err)
} else if _, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: c.Idx(), Query: `
Set(0, f=0)
@ -1981,9 +1979,9 @@ func TestExecutor_Execute_TopN(t *testing.T) {
// Set columns for rows 0, 10, & 20 across two shards.
if idx, err := hldr.CreateIndex(c.Idx(), pilosa.IndexOptions{Keys: true}); err != nil {
t.Fatal(err)
} else if _, err := idx.CreateField("f", pilosa.OptFieldTypeDefault(), pilosa.OptFieldKeys()); err != nil {
} else if _, err := idx.CreateField("f", pilosa.OptFieldKeys()); err != nil {
t.Fatal(err)
} else if _, err := idx.CreateField("other", pilosa.OptFieldTypeDefault(), pilosa.OptFieldKeys()); err != nil {
} else if _, err := idx.CreateField("other", pilosa.OptFieldKeys()); err != nil {
t.Fatal(err)
} else if _, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: c.Idx(), Query: `
Set("a", f="foo")
@ -2027,7 +2025,7 @@ func TestExecutor_Execute_TopN(t *testing.T) {
// Set data on the "f" field.
if idx, err := hldr.CreateIndex(c.Idx(), pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
} else if _, err := idx.CreateField("f", pilosa.OptFieldTypeDefault()); err != nil {
} else if _, err := idx.CreateField("f"); err != nil {
t.Fatal(err)
} else if _, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: c.Idx(), Query: `
Set(0, f=0)
@ -2327,7 +2325,7 @@ func TestExecutor_Execute_MinMax(t *testing.T) {
// This extra field exists to make there be shards which are present,
// but have no decimal values set, to make sure they don't break
// the results.
if _, err := idx.CreateFieldIfNotExists("z", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := idx.CreateFieldIfNotExists("z"); err != nil {
t.Fatal(err)
}
if res, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: c.Idx(), Query: `Set(1, z=0)`}); err != nil {
@ -2508,7 +2506,7 @@ func TestExecutor_Execute_MinMax(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("x", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := idx.CreateField("x"); err != nil {
t.Fatal(err)
}
@ -2572,7 +2570,7 @@ func TestExecutor_Execute_MinMax(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("x", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := idx.CreateField("x"); err != nil {
t.Fatal(err)
}
@ -2665,7 +2663,7 @@ func TestExecutor_Execute_MinMaxRow(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("f", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := idx.CreateField("f"); err != nil {
t.Fatal(err)
}
@ -2785,7 +2783,7 @@ func TestExecutor_Execute_Sum(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("x", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := idx.CreateField("x"); err != nil {
t.Fatal(err)
}
@ -2919,7 +2917,7 @@ func TestExecutor_Execute_Sum(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("x", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := idx.CreateField("x"); err != nil {
t.Fatal(err)
}
@ -3010,7 +3008,7 @@ func TestExecutor_Execute_Row_BSIGroup(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("f", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := idx.CreateField("f"); err != nil {
t.Fatal(err)
}
@ -3310,7 +3308,7 @@ func TestExecutor_Execute_Range_BSIGroup_Deprecated(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("f", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := idx.CreateField("f"); err != nil {
t.Fatal(err)
}
@ -3937,7 +3935,7 @@ func TestExecutor_Execute_Existence(t *testing.T) {
hldr := c.GetHolder(0)
index := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{TrackExistence: true})
_, err := index.CreateField("f", pilosa.OptFieldTypeDefault())
_, err := index.CreateField("f")
if err != nil {
t.Fatal(err)
}
@ -4339,7 +4337,7 @@ func TestExecutor_Execute_All(t *testing.T) {
defer c.Close()
hldr := c.GetHolder(0)
index := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{TrackExistence: true})
fld, err := index.CreateField("f", pilosa.OptFieldTypeDefault())
fld, err := index.CreateField("f")
if err != nil {
t.Fatal(err)
}
@ -4431,7 +4429,7 @@ func TestExecutor_Execute_All(t *testing.T) {
defer c.Close()
hldr := c.GetHolder(0)
index := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{TrackExistence: true, Keys: true})
fld, err := index.CreateField("f", pilosa.OptFieldTypeDefault())
fld, err := index.CreateField("f")
if err != nil {
t.Fatal(err)
}
@ -4496,7 +4494,7 @@ func TestExecutor_Execute_All(t *testing.T) {
defer c.Close()
hldr := c.GetHolder(0)
index := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{TrackExistence: true})
_, err := index.CreateField("f", pilosa.OptFieldTypeDefault())
_, err := index.CreateField("f")
if err != nil {
t.Fatal(err)
}
@ -4540,7 +4538,7 @@ func TestExecutor_Execute_ClearRow(t *testing.T) {
defer c.Close()
hldr := c.GetHolder(0)
index := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{TrackExistence: true})
_, err := index.CreateField("f", pilosa.OptFieldTypeDefault())
_, err := index.CreateField("f")
if err != nil {
t.Fatal(err)
}
@ -4620,10 +4618,10 @@ func TestExecutor_Execute_SetRow(t *testing.T) {
defer c.Close()
hldr := c.GetHolder(0)
index := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{TrackExistence: true})
if _, err := index.CreateField("f", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := index.CreateField("f"); err != nil {
t.Fatal(err)
}
if _, err := index.CreateField("tmp", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := index.CreateField("tmp"); err != nil {
t.Fatal(err)
}
@ -4675,7 +4673,7 @@ func TestExecutor_Execute_SetRow(t *testing.T) {
defer c.Close()
hldr := c.GetHolder(0)
idx := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{TrackExistence: true})
_, err := idx.CreateField("f", pilosa.OptFieldTypeDefault())
_, err := idx.CreateField("f")
if err != nil {
t.Fatal(err)
}
@ -4728,7 +4726,7 @@ func TestExecutor_Execute_SetRow(t *testing.T) {
defer c.Close()
hldr := c.GetHolder(0)
index := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{TrackExistence: true})
_, err := index.CreateField("f", pilosa.OptFieldTypeDefault())
_, err := index.CreateField("f")
if err != nil {
t.Fatal(err)
}
@ -4769,7 +4767,7 @@ func TestExecutor_Execute_SetRow(t *testing.T) {
defer c.Close()
hldr := c.GetHolder(0)
index := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{TrackExistence: true})
if _, err := index.CreateField("f", pilosa.OptFieldTypeDefault(), pilosa.OptFieldKeys()); err != nil {
if _, err := index.CreateField("f", pilosa.OptFieldKeys()); err != nil {
t.Fatal(err)
}
@ -4817,7 +4815,7 @@ func TestExecutor_Execute_SetRow(t *testing.T) {
func benchmarkExistence(nn bool, b *testing.B) {
c := test.MustUnsharedCluster(b, 1)
var err error
c.GetIdleNode(0).Config.DataDir, err = testhook.TempDirInDir(b, *TempDir, "benchmarkExistence")
c.GetIdleNode(0).Config.DataDir, err = testhook.TempDir(b, "benchmarkExistence")
if err != nil {
b.Fatalf("getting temp dir: %v", err)
}
@ -6445,7 +6443,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
func BenchmarkGroupBy(b *testing.B) {
c := test.MustUnsharedCluster(b, 1)
var err error
c.GetIdleNode(0).Config.DataDir, err = testhook.TempDirInDir(b, *TempDir, "benchmarkGroupBy-")
c.GetIdleNode(0).Config.DataDir, err = testhook.TempDir(b, "benchmarkGroupBy-")
if err != nil {
b.Fatalf("getting temp dir: %v", err)
}
@ -6634,7 +6632,7 @@ func TestExecutor_Execute_IncludesColumn(t *testing.T) {
cmd := c.GetNode(0)
hldr := c.GetHolder(0)
index := hldr.MustCreateIndexIfNotExists(c.Idx(), pilosa.IndexOptions{Keys: true})
if _, err := index.CreateField("general", pilosa.OptFieldTypeDefault(), pilosa.OptFieldKeys()); err != nil {
if _, err := index.CreateField("general", pilosa.OptFieldKeys()); err != nil {
t.Fatal(err)
}
@ -6705,7 +6703,7 @@ func TestExecutor_Execute_MinMaxCountEqual(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("x", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := idx.CreateField("x"); err != nil {
t.Fatal(err)
}

32
field.go
View file

@ -16,12 +16,11 @@ import (
"sync"
"time"
"github.com/featurebasedb/featurebase/v3/disco"
"github.com/featurebasedb/featurebase/v3/pql"
"github.com/featurebasedb/featurebase/v3/roaring"
"github.com/featurebasedb/featurebase/v3/stats"
"github.com/featurebasedb/featurebase/v3/testhook"
"github.com/featurebasedb/featurebase/v3/tracing"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/stats"
"github.com/molecula/featurebase/v3/testhook"
"github.com/molecula/featurebase/v3/tracing"
"github.com/pkg/errors"
)
@ -85,7 +84,6 @@ type Field struct {
broadcaster broadcaster
Stats stats.StatsClient
schemator disco.Schemator
serializer Serializer
// Field options.
@ -379,21 +377,6 @@ func OptFieldTypeBool() FieldOption {
}
}
// NewField returns a new instance of field.
// NOTE: This function is only used in tests, which is why
// it only takes a single `FieldOption` (the assumption being
// that it's of the type `OptFieldType*`). This means
// this function couldn't be used to set, for example,
// `FieldOptions.Keys`.
func NewField(holder *Holder, path, index, name string, opts FieldOption) (*Field, error) {
err := ValidateName(name)
if err != nil {
return nil, errors.Wrap(err, "validating name")
}
return newField(holder, path, index, name, opts)
}
// newField returns a new instance of field (without name validation).
func newField(holder *Holder, path, index, name string, opts FieldOption) (*Field, error) {
// Apply functional option.
@ -413,7 +396,6 @@ func newField(holder *Holder, path, index, name string, opts FieldOption) (*Fiel
broadcaster: NopBroadcaster,
Stats: stats.NopStatsClient,
schemator: disco.NopSchemator,
serializer: NopSerializer,
options: applyDefaultOptions(&fo),
@ -1197,7 +1179,7 @@ func (f *Field) deleteView(name string) error {
}
// Delete the view from etcd as the system of record.
if err := f.schemator.DeleteView(context.TODO(), f.index, f.name, name); err != nil {
if err := f.holder.Schemator.DeleteView(context.TODO(), f.index, f.name, name); err != nil {
return errors.Wrapf(err, "deleting view from etcd: %s/%s/%s", f.index, f.name, name)
}
@ -2332,7 +2314,7 @@ func (f *Field) persistView(ctx context.Context, cvm *CreateViewMessage) error {
return ErrViewRequired
}
return f.schemator.CreateView(ctx, cvm.Index, cvm.Field, cvm.View)
return f.holder.Schemator.CreateView(ctx, cvm.Index, cvm.Field, cvm.View)
}
// Timestamp field ranges.

166
field_internal_test.go
View file

@ -6,18 +6,16 @@ import (
"context"
"fmt"
"math"
"os"
"reflect"
"strconv"
"strings"
"testing"
"time"
"github.com/featurebasedb/featurebase/v3/pql"
"github.com/featurebasedb/featurebase/v3/roaring"
"github.com/featurebasedb/featurebase/v3/shardwidth"
"github.com/featurebasedb/featurebase/v3/testhook"
. "github.com/featurebasedb/featurebase/v3/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/shardwidth"
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
)
// CorruptAMutex breaks a mutex in order to test the mutex-corruption stuff.
@ -184,7 +182,7 @@ func TestBSIGroup_BaseValue(t *testing.T) {
}
func TestField_ValCountize(t *testing.T) {
f := OpenField(t, OptFieldTypeDefault())
_, _, f := newTestField(t)
// check that you get an empty val count and err
// BSIGroupNotFound on nil bsig from
// f.bsiGroup(f.name)
@ -201,7 +199,7 @@ func TestField_ValCountize(t *testing.T) {
// Ensure field can open and retrieve a view.
func TestField_DeleteView(t *testing.T) {
f := OpenField(t, OptFieldTypeDefault())
_, _, f := newTestField(t)
viewName := viewStandard + "_v"
@ -231,93 +229,47 @@ func TestField_DeleteView(t *testing.T) {
}
}
// TestField represents a test wrapper for Field.
type TestField struct {
*Field
parent *Index
tb testing.TB
}
// NewTestField returns a new instance of TestField d/0.
func NewTestField(t testing.TB, opts FieldOption) *TestField {
path, err := testhook.TempDirInDir(t, *TempDir, "pilosa-field-")
// reopenTestField closes the field's parent index, then
// reopens it using its cached schema, and returns the corresponding
// field data structure from the reopened index.
func reopenTestField(t testing.TB, f *Field) (*Field, error) {
name := f.Name()
if err := f.idx.Close(); err != nil {
f.idx = nil
return nil, err
}
schema, err := f.holder.Schemator.Schema(context.Background())
if err != nil {
t.Fatal(err)
return nil, err
}
cfg := DefaultHolderConfig()
cfg.StorageConfig.FsyncEnabled = false
cfg.RBFConfig.FsyncEnabled = false
h := NewHolder(path, cfg)
PanicOn(h.Open())
idx, err := h.CreateIndex("i", IndexOptions{})
if err != nil {
panic(err)
if err := f.idx.OpenWithSchema(schema[f.idx.name]); err != nil {
f.idx = nil
return nil, err
}
field, err := idx.CreateField("f", opts)
if err != nil {
t.Fatal(err)
}
tf := &TestField{Field: field, parent: idx, tb: t}
testhook.Cleanup(t, func() {
h.Close()
})
return tf
return f.idx.Field(name), nil
}
// OpenField returns a new, opened field at a temporary path.
func OpenField(t testing.TB, opts FieldOption) *TestField {
f := NewTestField(t, opts)
return f
}
// Close closes the field and removes the underlying data.
func (f *TestField) Close() error {
if f.idx != nil {
PanicOn(f.idx.holder.txf.CloseIndex(f.idx))
}
defer os.RemoveAll(f.Path())
return f.Field.Close()
}
// Reopen closes the index and reopens it.
func (f *TestField) Reopen() error {
name := f.Field.Name()
if err := f.parent.Close(); err != nil {
f.parent = nil
return err
}
schema, err := f.parent.Schemator.Schema(context.Background())
if err != nil {
return err
}
if err := f.parent.OpenWithSchema(schema[f.parent.name]); err != nil {
f.parent = nil
return err
}
f.Field = f.parent.Field(name)
return nil
}
func (f *TestField) MustSetBit(qcx *Qcx, row, col uint64, ts ...time.Time) {
// testFieldSetBit sets a bit and checks for an error, using a provided qcx and an
// optional timestamp or series of timestamps. if multiple times are provided,
// the underlying set bit operation is repeated for all of them.
func testFieldSetBit(tb testing.TB, qcx *Qcx, f *Field, row, col uint64, ts ...time.Time) {
if len(ts) == 0 {
_, err := f.Field.SetBit(qcx, row, col, nil)
_, err := f.SetBit(qcx, row, col, nil)
if err != nil {
panic(err)
tb.Fatalf("setting bit: %v", err)
}
}
for _, t := range ts {
_, err := f.Field.SetBit(qcx, row, col, &t)
_, err := f.SetBit(qcx, row, col, &t)
if err != nil {
panic(err)
tb.Fatalf("setting bit: %v", err)
}
}
}
// Ensure field can open and retrieve a view.
func TestField_CreateViewIfNotExists(t *testing.T) {
f := OpenField(t, OptFieldTypeDefault())
_, _, f := newTestField(t)
// Create view.
view, err := f.createViewIfNotExists("v")
@ -341,7 +293,7 @@ func TestField_CreateViewIfNotExists(t *testing.T) {
}
func TestField_SetTimeQuantum(t *testing.T) {
f := OpenField(t, OptFieldTypeTime(TimeQuantum("YMDH"), "0"))
_, _, f := newTestField(t, OptFieldTypeTime(TimeQuantum("YMDH"), "0"))
// Retrieve time quantum.
if q := f.TimeQuantum(); q != TimeQuantum("YMDH") {
@ -349,7 +301,8 @@ func TestField_SetTimeQuantum(t *testing.T) {
}
// Reload field and verify that it is persisted.
if err := f.Reopen(); err != nil {
f, err := reopenTestField(t, f)
if err != nil {
t.Fatal(err)
} else if q := f.TimeQuantum(); q != TimeQuantum("YMDH") {
t.Fatalf("unexpected quantum (reopen): %s", q)
@ -357,27 +310,27 @@ func TestField_SetTimeQuantum(t *testing.T) {
}
func TestField_RowTime(t *testing.T) {
f := OpenField(t, OptFieldTypeTime(TimeQuantum("YMDH"), "0"))
_, _, f := newTestField(t, OptFieldTypeTime(TimeQuantum("YMDH"), "0"))
// Obtain transaction.
qcx := f.idx.Txf().NewWritableQcx()
qcx := f.holder.Txf().NewWritableQcx()
defer qcx.Abort()
f.MustSetBit(qcx, 1, 1, time.Date(2010, time.January, 5, 12, 0, 0, 0, time.UTC))
f.MustSetBit(qcx, 1, 2, time.Date(2011, time.January, 5, 12, 0, 0, 0, time.UTC))
f.MustSetBit(qcx, 1, 3, time.Date(2010, time.February, 5, 12, 0, 0, 0, time.UTC))
f.MustSetBit(qcx, 1, 4, time.Date(2010, time.January, 6, 12, 0, 0, 0, time.UTC))
f.MustSetBit(qcx, 1, 5, time.Date(2010, time.January, 5, 13, 0, 0, 0, time.UTC))
testFieldSetBit(t, qcx, f, 1, 1, time.Date(2010, time.January, 5, 12, 0, 0, 0, time.UTC))
testFieldSetBit(t, qcx, f, 1, 2, time.Date(2011, time.January, 5, 12, 0, 0, 0, time.UTC))
testFieldSetBit(t, qcx, f, 1, 3, time.Date(2010, time.February, 5, 12, 0, 0, 0, time.UTC))
testFieldSetBit(t, qcx, f, 1, 4, time.Date(2010, time.January, 6, 12, 0, 0, 0, time.UTC))
testFieldSetBit(t, qcx, f, 1, 5, time.Date(2010, time.January, 5, 13, 0, 0, 0, time.UTC))
// Warning: Right now this is misleading, and doesn't really do anything. We
// already committed each change as we got there. SOME DAY we will fix this.
PanicOn(qcx.Finish())
qcx = f.idx.Txf().NewQcx()
qcx = f.holder.Txf().NewQcx()
defer qcx.Abort()
// obtain 2nd transaction to read it back.
qcx = f.idx.Txf().NewQcx()
qcx = f.holder.Txf().NewQcx()
defer qcx.Abort()
if r, err := f.RowTime(qcx, 1, time.Date(2010, time.November, 5, 12, 0, 0, 0, time.UTC), "Y"); err != nil {
@ -414,7 +367,7 @@ func TestField_RowTime(t *testing.T) {
func TestField_PersistAvailableShards(t *testing.T) {
availableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write
f := OpenField(t, OptFieldTypeDefault())
_, _, f := newTestField(t)
// bm represents remote available shards.
bm := roaring.NewBitmap(1, 2, 3)
@ -425,7 +378,8 @@ func TestField_PersistAvailableShards(t *testing.T) {
time.Sleep(2 * availableShardFileFlushDuration.Get())
// Reload field and verify that shard data is persisted.
if err := f.Reopen(); err != nil {
f, err := reopenTestField(t, f)
if err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(f.protectedRemoteAvailableShards().Slice(), bm.Slice()) {
t.Fatalf("unexpected available shards (reopen). expected: %v, but got: %v", bm.Slice(), f.protectedRemoteAvailableShards().Slice())
@ -500,7 +454,7 @@ func TestField_ApplyOptions(t *testing.T) {
// into consideration. This would cause an import of 1/8/1
// to result in a value of 9 instead of 1.
func TestBSIGroup_importValue(t *testing.T) {
f := OpenField(t, OptFieldTypeInt(-100, 200))
_, _, f := newTestField(t, OptFieldTypeInt(-100, 200))
qcx := f.idx.holder.txf.NewQcx()
defer qcx.Abort()
@ -548,7 +502,7 @@ func TestBSIGroup_importValue(t *testing.T) {
// benchmarkImportValues is a helper function to explore, very roughly, the cost
// of setting values using the special setter used for imports.
func benchmarkFieldImportValues(b *testing.B, qcx *Qcx, bitDepth uint64, f *TestField, cfunc func(uint64) uint64) {
func benchmarkFieldImportValues(b *testing.B, qcx *Qcx, bitDepth uint64, f *Field, cfunc func(uint64) uint64) {
batches := makeBenchmarkImportValueData(b, bitDepth, cfunc)
for _, req := range batches {
// NOTE: We assume everything's in Shard 0 for now.
@ -564,7 +518,7 @@ func BenchmarkField_ImportValue(b *testing.B) {
depths := []uint64{4, 8, 16, 32}
for _, bitDepth := range depths {
f := OpenField(b, OptFieldTypeInt(0, 1<<bitDepth))
_, _, f := newTestField(b, OptFieldTypeInt(0, 1<<bitDepth))
qcx := f.idx.holder.txf.NewQcx()
defer qcx.Abort()
@ -579,7 +533,7 @@ func BenchmarkField_ImportValue(b *testing.B) {
}
func TestIntField_MinMaxForShard(t *testing.T) {
f := OpenField(t, OptFieldTypeInt(-100, 200))
_, _, f := newTestField(t, OptFieldTypeInt(-100, 200))
qcx := f.idx.holder.txf.NewQcx()
defer qcx.Abort()
@ -721,7 +675,7 @@ func TestDecimalField_MinMaxBoundaries(t *testing.T) {
},
} {
t.Run("minmax"+strconv.Itoa(i), func(t *testing.T) {
_, err := NewField(th, "no-path", "i", "f", OptFieldTypeDecimal(test.scale, test.min, test.max))
_, err := newField(th, "no-path", "i", "f", OptFieldTypeDecimal(test.scale, test.min, test.max))
if err != nil && test.expErr {
if !strings.Contains(err.Error(), "is not supported") {
t.Fatal(err)
@ -736,7 +690,7 @@ func TestDecimalField_MinMaxBoundaries(t *testing.T) {
}
func TestDecimalField_MinMaxForShard(t *testing.T) {
f := OpenField(t, OptFieldTypeDecimal(3))
_, _, f := newTestField(t, OptFieldTypeDecimal(3))
qcx := f.idx.holder.txf.NewQcx()
defer qcx.Abort()
@ -810,7 +764,7 @@ func TestDecimalField_MinMaxForShard(t *testing.T) {
}
func TestBSIGroup_TxReopenDB(t *testing.T) {
f := OpenField(t, OptFieldTypeInt(-100, 200))
_, _, f := newTestField(t, OptFieldTypeInt(-100, 200))
qcx := f.idx.holder.txf.NewQcx()
defer qcx.Abort()
@ -857,19 +811,22 @@ func TestBSIGroup_TxReopenDB(t *testing.T) {
} // loop
// the test: can we re-open a BSI fragment under Tx store
_ = f.Reopen()
f, err := reopenTestField(t, f)
if err != nil {
t.Fatalf("reopening test field: %v", err)
}
}
// Ensure that an integer field has the same BitDepth after reopening.
func TestField_SaveMeta(t *testing.T) {
f := OpenField(t, OptFieldTypeInt(-10, 1000))
_, _, f := newTestField(t, OptFieldTypeInt(-10, 1000))
colID := uint64(1)
val := int64(88)
expBitDepth := uint64(7)
// Obtain transaction.
qcx := f.idx.Txf().NewWritableQcx()
qcx := f.holder.Txf().NewWritableQcx()
defer qcx.Abort()
if changed, err := f.SetValue(qcx, colID, val); err != nil {
@ -895,11 +852,12 @@ func TestField_SaveMeta(t *testing.T) {
}
// Reload field and verify that it is persisted.
if err := f.Reopen(); err != nil {
f, err := reopenTestField(t, f)
if err != nil {
t.Fatal(err)
}
qcx = f.idx.Txf().NewQcx()
qcx = f.holder.Txf().NewQcx()
defer qcx.Abort()
if f.options.BitDepth != expBitDepth {
@ -916,7 +874,7 @@ func TestField_SaveMeta(t *testing.T) {
}
func TestFieldViewsByTimeRange(t *testing.T) {
f := OpenField(t, OptFieldTypeTime("YMD", "0", false))
_, _, f := newTestField(t, OptFieldTypeTime("YMD", "0", false))
for _, date := range []string{
// a handful of YMD parameters describing dates that we could have data for
"2021",

57
field_test.go
View file

@ -15,20 +15,24 @@ import (
"github.com/featurebasedb/featurebase/v3/test"
"github.com/featurebasedb/featurebase/v3/testhook"
"github.com/google/go-cmp/cmp"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/test"
"github.com/pkg/errors"
)
// Ensure a field can set & read a bsiGroup value.
func TestField_SetValue(t *testing.T) {
t.Run("OK", func(t *testing.T) {
idx := test.MustOpenIndex(t)
h, idx := test.MustOpenIndex(t)
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
if err != nil {
t.Fatal(err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := h.Txf().NewWritableQcx()
defer qcx.Abort()
// You're going to note the lack of any commits here. That's
// because, when you have a writable Qcx, *every individual
@ -62,14 +66,14 @@ func TestField_SetValue(t *testing.T) {
})
t.Run("Overwrite", func(t *testing.T) {
idx := test.MustOpenIndex(t)
h, idx := test.MustOpenIndex(t)
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
if err != nil {
t.Fatal(err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := h.Txf().NewWritableQcx()
defer qcx.Abort()
// Set value.
@ -97,14 +101,14 @@ func TestField_SetValue(t *testing.T) {
})
t.Run("ErrBSIGroupNotFound", func(t *testing.T) {
idx := test.MustOpenIndex(t)
h, idx := test.MustOpenIndex(t)
f, err := idx.CreateField("f", pilosa.OptFieldTypeDefault())
f, err := idx.CreateField("f")
if err != nil {
t.Fatal(err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := h.Txf().NewWritableQcx()
defer qcx.Abort()
// Set value.
@ -114,13 +118,13 @@ func TestField_SetValue(t *testing.T) {
})
t.Run("ErrBSIGroupValueTooLow", func(t *testing.T) {
idx := test.MustOpenIndex(t)
h, idx := test.MustOpenIndex(t)
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(20, 30))
if err != nil {
t.Fatal(err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := h.Txf().NewWritableQcx()
defer qcx.Abort()
// Set value.
if _, err := f.SetValue(qcx, 100, 15); !errors.Is(err, pilosa.ErrBSIGroupValueTooLow) {
@ -129,14 +133,14 @@ func TestField_SetValue(t *testing.T) {
})
t.Run("ErrBSIGroupValueTooHigh", func(t *testing.T) {
idx := test.MustOpenIndex(t)
h, idx := test.MustOpenIndex(t)
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(20, 30))
if err != nil {
t.Fatal(err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := h.Txf().NewWritableQcx()
defer qcx.Abort()
// Set value.
@ -146,17 +150,6 @@ func TestField_SetValue(t *testing.T) {
})
}
func TestField_NameRestriction(t *testing.T) {
path, err := testhook.TempDir(t, "pilosa-field-")
if err != nil {
panic(err)
}
field, err := pilosa.NewField(pilosa.NewHolder(path, mustHolderConfig()), path, "i", ".meta", pilosa.OptFieldTypeDefault())
if field != nil {
t.Fatalf("unexpected field name %s", err)
}
}
// Ensure that field name validation is consistent.
func TestField_NameValidation(t *testing.T) {
validFieldNames := []string{
@ -176,21 +169,19 @@ func TestField_NameValidation(t *testing.T) {
"abc def",
"camelCase",
"UPPERCASE",
".meta",
"charact23112345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901",
}
path, err := testhook.TempDir(t, "pilosa-field-")
if err != nil {
panic(err)
}
_, idx := test.MustOpenIndex(t)
for _, name := range validFieldNames {
_, err := pilosa.NewField(pilosa.NewHolder(path, mustHolderConfig()), path, "i", name, pilosa.OptFieldTypeDefault())
_, err := idx.CreateField(name)
if err != nil {
t.Fatalf("unexpected field name: %s %s", name, err)
}
}
for _, name := range invalidFieldNames {
_, err := pilosa.NewField(pilosa.NewHolder(path, mustHolderConfig()), path, "i", name, pilosa.OptFieldTypeDefault())
_, err := idx.CreateField(name)
if err == nil {
t.Fatalf("expected error on field name: %s", name)
}
@ -201,14 +192,14 @@ const includeRemote = false // for calls to Index.AvailableShards(localOnly bool
// Ensure can update and delete available shards.
func TestField_AvailableShards(t *testing.T) {
idx := test.MustOpenIndex(t)
h, idx := test.MustOpenIndex(t)
f, err := idx.CreateField("fld-shards", pilosa.OptFieldTypeDefault())
f, err := idx.CreateField("fld-shards")
if err != nil {
t.Fatal(err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := h.Txf().NewWritableQcx()
defer qcx.Abort()
// Set values on shards 0 & 2, and verify.
@ -242,13 +233,13 @@ func TestField_AvailableShards(t *testing.T) {
func TestField_ClearValue(t *testing.T) {
t.Run("OK", func(t *testing.T) {
idx := test.MustOpenIndex(t)
h, idx := test.MustOpenIndex(t)
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
if err != nil {
t.Fatal(err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := h.Txf().NewWritableQcx()
defer qcx.Abort()
// Set value on field.

42
fragment.go
View file

@ -69,33 +69,13 @@ const (
bsiExistsBit = 0
bsiSignBit = 1
bsiOffsetBit = 2
// Roaring bitmap flags.
roaringFlagBSIv2 = 0x01 // indicates version using low bit for existence
)
// fragSpec saves a ton of duplicated strings for
// path, and index, field, view name strings.
type fragSpec struct {
index *Index
field *Field
fieldstr string
view *view
}
func (f *fragment) index() string {
return f.idx.name
}
func (f *fragment) field() string {
// initialization of _field *Field can
// go missing during tests that do incomplete setup.
// Hence we must keep fieldstr as a backup.
if f.fld == nil {
return f.fieldstr
} else {
f.fieldstr = ""
}
return f.fld.name
}
@ -113,9 +93,8 @@ type fragment struct {
// We save 20GB worth strings on some data sets by not duplicating
// the path, index, field, view strings on every fragment.
// Instead assemble strings on demand in field(), view(), path(), index().
fld *Field
fieldstr string
_view *view
fld *Field
_view *view
shard uint64
@ -151,19 +130,18 @@ type fragment struct {
}
// newFragment returns a new instance of fragment.
func newFragment(holder *Holder, spec fragSpec, shard uint64, flags byte) *fragment {
idx := holder.Index(spec.index.name)
func newFragment(holder *Holder, idx *Index, fld *Field, vw *view, shard uint64) *fragment {
checkIdx := holder.Index(idx.name)
if idx == nil {
vprint.PanicOn(fmt.Sprintf("got nil idx back for '%v' from holder!", spec.index))
if checkIdx == nil {
vprint.PanicOn(fmt.Sprintf("got nil idx back for '%v' from holder!", idx.Name()))
}
f := &fragment{
_view: spec.view,
fieldstr: spec.fieldstr,
fld: spec.field,
shard: shard,
idx: idx,
_view: vw,
fld: fld,
shard: shard,
idx: idx,
CacheType: DefaultCacheType,
CacheSize: DefaultCacheSize,

373
fragment_internal_test.go
View file

File diff suppressed because it is too large Load diff

59
holder.go
View file

@ -65,7 +65,7 @@ type Holder struct {
opened lockedChan
broadcaster broadcaster
schemator disco.Schemator
Schemator disco.Schemator
sharder disco.Sharder
serializer Serializer
@ -174,14 +174,14 @@ type lockedChan struct {
func (lc *lockedChan) Close() {
lc.mu.RLock()
defer lc.mu.RUnlock()
close(lc.ch)
lc.mu.RUnlock()
}
func (lc *lockedChan) Recv() {
lc.mu.RLock()
defer lc.mu.RUnlock()
<-lc.ch
lc.mu.RUnlock()
}
// HolderConfig holds configuration details that need to be set up at
@ -209,6 +209,10 @@ type HolderConfig struct {
LookupDBDSN string
}
// DefaultHolderConfig provides a holder config with reasonable
// defaults. Note that a production server would almost certainly
// need to override these; that's usually handled by server options
// such as OptServerOpenTranslateStore.
func DefaultHolderConfig() *HolderConfig {
return &HolderConfig{
PartitionN: disco.DefaultPartitionN,
@ -228,6 +232,20 @@ func DefaultHolderConfig() *HolderConfig {
}
}
// TestHolderConfig provides a holder config with reasonable
// defaults for tests. This means it tries to disable fsync
// and sets significantly smaller file size limits for RBF,
// for instance. Do not use this outside of the test
// infrastructure.
func TestHolderConfig() *HolderConfig {
cfg := DefaultHolderConfig()
cfg.StorageConfig.FsyncEnabled = false
cfg.RBFConfig.FsyncEnabled = false
cfg.RBFConfig.MaxSize = (1 << 28)
cfg.RBFConfig.MaxWALSize = (1 << 28)
return cfg
}
// NewHolder returns a new instance of Holder for the given path.
func NewHolder(path string, cfg *HolderConfig) *Holder {
if cfg == nil {
@ -258,7 +276,7 @@ func NewHolder(path string, cfg *HolderConfig) *Holder {
translationSyncer: cfg.TranslationSyncer,
serializer: cfg.Serializer,
sharder: cfg.Sharder,
schemator: cfg.Schemator,
Schemator: cfg.Schemator,
Logger: cfg.Logger,
Opts: HolderOpts{StorageBackend: cfg.StorageConfig.Backend},
@ -295,7 +313,7 @@ func (h *Holder) deletePerShard(index *Index, shard uint64) error {
return nil
}
tx := index.Txf().NewTx(Txo{Write: !writable, Index: index, Shard: shard})
tx := h.Txf().NewTx(Txo{Write: !writable, Index: index, Shard: shard})
defer tx.Rollback()
// filter rows based on having _exists>=1, which is used to flag delete in-flight
@ -395,7 +413,7 @@ func (h *Holder) Open() error {
}
// Load schema from etcd.
schema, err := h.schemator.Schema(context.Background())
schema, err := h.Schemator.Schema(context.Background())
if err != nil {
return errors.Wrap(err, "getting schema")
}
@ -644,9 +662,9 @@ func (h *Holder) limitedSchema() ([]*IndexInfo, error) {
}
func (h *Holder) schema(ctx context.Context, includeViews bool) ([]*IndexInfo, error) {
schema, err := h.schemator.Schema(ctx)
schema, err := h.Schemator.Schema(ctx)
if err != nil {
return nil, errors.Wrapf(err, "getting schema via schemator")
return nil, errors.Wrapf(err, "getting schema via Schemator")
}
a := make([]*IndexInfo, 0, len(schema))
@ -784,7 +802,7 @@ func (h *Holder) CreateIndex(name string, opt IndexOptions) (*Index, error) {
// latest schema from etcd.
type LoadSchemaMessage struct{}
// LoadSchema creates all indexes based on the information stored in schemator.
// LoadSchema creates all indexes based on the information stored in Schemator.
// It does not return an error if an index already exists. The thinking is that
// this method will load all indexes that don't already exist. We likely want to
// revisit this; for example, we might want to confirm that the createdAt
@ -796,7 +814,7 @@ func (h *Holder) LoadSchema() error {
return h.loadSchema()
}
// LoadIndex creates an index based on the information stored in schemator.
// LoadIndex creates an index based on the information stored in Schemator.
// An error is returned if the index already exists.
func (h *Holder) LoadIndex(name string) (*Index, error) {
h.mu.Lock()
@ -809,7 +827,7 @@ func (h *Holder) LoadIndex(name string) (*Index, error) {
return h.loadIndex(name)
}
// LoadField creates a field based on the information stored in schemator.
// LoadField creates a field based on the information stored in Schemator.
// An error is returned if the field already exists.
func (h *Holder) LoadField(index, field string) (*Field, error) {
// Ensure field doesn't already exist.
@ -823,7 +841,7 @@ func (h *Holder) LoadField(index, field string) (*Field, error) {
return h.loadField(index, field)
}
// LoadView creates a view based on the information stored in schemator. Unlike
// LoadView creates a view based on the information stored in Schemator. Unlike
// index and field, it is not considered an error if the view already exists.
func (h *Holder) LoadView(index, field, view string) (*view, error) {
// If the view already exists, just return with it here.
@ -893,7 +911,7 @@ func (h *Holder) persistIndex(ctx context.Context, cim *CreateIndexMessage) erro
if b, err := h.serializer.Marshal(cim); err != nil {
return errors.Wrap(err, "marshaling")
} else if err := h.schemator.CreateIndex(ctx, cim.Index, b); err != nil {
} else if err := h.Schemator.CreateIndex(ctx, cim.Index, b); err != nil {
return errors.Wrapf(err, "writing index to disco: %s", cim.Index)
}
return nil
@ -938,7 +956,7 @@ func (h *Holder) createIndex(cim *CreateIndexMessage, broadcast bool) (*Index, e
}
func (h *Holder) loadSchema() error {
schema, err := h.schemator.Schema(context.TODO())
schema, err := h.Schemator.Schema(context.TODO())
if err != nil {
return errors.Wrap(err, "getting schema")
}
@ -946,7 +964,7 @@ func (h *Holder) loadSchema() error {
// TODO: This is kind of inefficient because we're ignoring the index.Data
// and field.Data values, which contains the index and field information,
// and only using the map key to call loadIndex() and loadField(). These
// make another call to schemator to get the same index and field
// make another call to Schemator to get the same index and field
// information that we already have in the map. It probably makes sense to
// either copy the parts of the loadIndex and loadField methods here (like
// decodeCreateIndexMessage) or split loadIndex and loadField into smaller
@ -974,7 +992,7 @@ func (h *Holder) loadSchema() error {
}
func (h *Holder) loadIndex(indexName string) (*Index, error) {
b, err := h.schemator.Index(context.TODO(), indexName)
b, err := h.Schemator.Index(context.TODO(), indexName)
if err != nil {
return nil, errors.Wrapf(err, "getting index: %s", indexName)
}
@ -988,7 +1006,7 @@ func (h *Holder) loadIndex(indexName string) (*Index, error) {
}
func (h *Holder) loadField(indexName, fieldName string) (*Field, error) {
b, err := h.schemator.Field(context.TODO(), indexName, fieldName)
b, err := h.Schemator.Field(context.TODO(), indexName, fieldName)
if err != nil {
return nil, errors.Wrapf(err, "getting field: %s/%s", indexName, fieldName)
}
@ -1008,7 +1026,7 @@ func (h *Holder) loadField(indexName, fieldName string) (*Field, error) {
}
func (h *Holder) loadView(indexName, fieldName, viewName string) (*view, error) {
b, err := h.schemator.View(context.Background(), indexName, fieldName, viewName)
b, err := h.Schemator.View(context.Background(), indexName, fieldName, viewName)
if err != nil {
return nil, errors.Wrapf(err, "getting view: %s/%s/%s", indexName, fieldName, viewName)
} else if !b {
@ -1032,7 +1050,6 @@ func (h *Holder) newIndex(path, name string) (*Index, error) {
index.Stats = h.Stats.WithTags(fmt.Sprintf("index:%s", index.Name()))
index.broadcaster = h.broadcaster
index.serializer = h.serializer
index.Schemator = h.schemator
index.OpenTranslateStore = h.OpenTranslateStore
index.translationSyncer = h.translationSyncer
return index, nil
@ -1050,7 +1067,7 @@ func (h *Holder) DeleteIndex(name string) error {
}
// Delete the index from etcd as the system of record.
if err := h.schemator.DeleteIndex(context.TODO(), name); err != nil {
if err := h.Schemator.DeleteIndex(context.TODO(), name); err != nil {
return errors.Wrapf(err, "deleting index from etcd: %s", name)
}
@ -1061,7 +1078,7 @@ func (h *Holder) DeleteIndex(name string) error {
// remove any backing store.
if err := h.txf.DeleteIndex(name); err != nil {
return errors.Wrap(err, "index.Txf.DeleteIndex")
return errors.Wrap(err, "h.Txf.DeleteIndex")
}
// Delete index directory.

30
holder_internal_test.go
View file

@ -4,33 +4,20 @@ package pilosa
import (
"testing"
"github.com/featurebasedb/featurebase/v3/disco"
"github.com/featurebasedb/featurebase/v3/testhook"
)
// mustHolderConfig sets up a default holder config for tests.
func mustHolderConfig() *HolderConfig {
cfg := DefaultHolderConfig()
cfg.StorageConfig.FsyncEnabled = false
cfg.RBFConfig.FsyncEnabled = false
cfg.Schemator = disco.NewInMemSchemator()
cfg.Sharder = disco.InMemSharder
return cfg
}
func setupTest(t *testing.T, h *Holder, rowCol []rowCols, indexName string) (*Index, *Field) {
idx, err := h.CreateIndexIfNotExists(indexName, IndexOptions{TrackExistence: true})
if err != nil {
t.Fatalf("failed to create index %v: %v", indexName, err)
}
f, err := idx.CreateFieldIfNotExists("f", OptFieldTypeDefault())
f, err := idx.CreateFieldIfNotExists("f")
if err != nil {
t.Fatalf("failed to create field in index %v: %v", indexName, err)
}
existencefield := idx.existenceFld
qcx := idx.Txf().NewWritableQcx()
qcx := h.Txf().NewWritableQcx()
defer qcx.Abort()
for _, r := range rowCol {
@ -62,14 +49,7 @@ type rowCols struct {
}
func TestHolder_ProcessDeleteInflight(t *testing.T) {
path, _ := testhook.TempDir(t, "delete-inflight")
h := NewHolder(path, mustHolderConfig())
defer h.Close()
err := h.Open()
if err != nil {
t.Fatalf("failed to open holder: %v", err)
}
h := newTestHolder(t)
rowCol := []rowCols{
{1, 1},
@ -81,7 +61,7 @@ func TestHolder_ProcessDeleteInflight(t *testing.T) {
idx1, f1 := setupTest(t, h, rowCol, "idxdelete1")
idx2, f2 := setupTest(t, h, rowCol, "idxdelete2")
err = h.processDeleteInflight()
err := h.processDeleteInflight()
if err != nil {
t.Fatalf("failed to delete: %v", err)
}
@ -97,7 +77,7 @@ func TestHolder_ProcessDeleteInflight(t *testing.T) {
for _, test := range tests {
func() {
idx, f := test.idx, test.f
qcx := idx.Txf().NewQcx()
qcx := h.Txf().NewQcx()
defer qcx.Abort()
for _, r := range rowCol {
row, err := f.Row(qcx, r.row)

35
holder_test.go
View file

@ -11,30 +11,24 @@ import (
"testing"
"time"
pilosa "github.com/featurebasedb/featurebase/v3"
"github.com/featurebasedb/featurebase/v3/disco"
"github.com/featurebasedb/featurebase/v3/pql"
"github.com/featurebasedb/featurebase/v3/test"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/test"
"github.com/pkg/errors"
)
// mustHolderConfig provides a default test-friendly holder config.
func mustHolderConfig() *pilosa.HolderConfig {
cfg := pilosa.DefaultHolderConfig()
cfg.StorageConfig.Backend = "rbf"
cfg.StorageConfig.FsyncEnabled = false
cfg.RBFConfig.FsyncEnabled = false
cfg.Schemator = disco.InMemSchemator
cfg.Sharder = disco.InMemSharder
return cfg
}
func TestHolder_Open(t *testing.T) {
t.Run("ErrIndexPermission", func(t *testing.T) {
if os.Geteuid() == 0 {
t.Skip("Skipping permissions test since user is root.")
}
h := test.MustOpenHolder(t)
// Manual open because MustOpenHolder closes automatically and fails the test on
// double-close.
h := test.NewHolder(t)
err := h.Open()
if err != nil {
t.Fatalf("opening holder: %v", err)
}
// no automatic close here, because we manually close this, and then
// *fail* to reopen it.
@ -56,7 +50,6 @@ func TestHolder_Open(t *testing.T) {
t.Run("ForeignIndex", func(t *testing.T) {
t.Run("ErrForeignIndexNotFound", func(t *testing.T) {
h := test.MustOpenHolder(t)
defer h.Close()
if idx, err := h.CreateIndex("foo", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
@ -73,7 +66,6 @@ func TestHolder_Open(t *testing.T) {
// Foreign index zzz is opened after foo/bar.
t.Run("ForeignIndexNotOpenYet", func(t *testing.T) {
h := test.MustOpenHolder(t)
defer h.Close()
if _, err := h.CreateIndex("zzz", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
@ -93,7 +85,6 @@ func TestHolder_Open(t *testing.T) {
// Foreign index aaa is opened before foo/bar.
t.Run("ForeignIndexIsOpen", func(t *testing.T) {
h := test.MustOpenHolder(t)
defer h.Close()
if _, err := h.CreateIndex("aaa", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
@ -113,7 +104,6 @@ func TestHolder_Open(t *testing.T) {
// Try to re-create existing index
t.Run("CreateIndexIfNotExists", func(t *testing.T) {
h := test.MustOpenHolder(t)
defer h.Close()
idx1, err := h.CreateIndexIfNotExists("aaa", pilosa.IndexOptions{})
if err != nil {
@ -141,7 +131,6 @@ func TestHolder_Open(t *testing.T) {
func TestHolder_HasData(t *testing.T) {
t.Run("IndexDirectory", func(t *testing.T) {
h := test.MustOpenHolder(t)
defer h.Close()
if ok, err := h.HasData(); ok || err != nil {
t.Fatal("expected HasData to return false, no err, but", ok, err)
@ -158,7 +147,6 @@ func TestHolder_HasData(t *testing.T) {
t.Run("Peek", func(t *testing.T) {
h := test.MustOpenHolder(t)
defer h.Close()
if ok, err := h.HasData(); ok || err != nil {
t.Fatal("expected HasData to return false, no err, but", ok, err)
@ -180,7 +168,7 @@ func TestHolder_HasData(t *testing.T) {
// Note that we are intentionally not using test.NewHolder,
// because we want to create a Holder object with an invalid path,
// rather than creating a valid holder with a temporary path.
h := pilosa.NewHolder("bad-path", mustHolderConfig())
h := pilosa.NewHolder("bad-path", pilosa.TestHolderConfig())
if ok, err := h.HasData(); ok || err != nil {
t.Fatal("expected HasData to return false, no err, but", ok, err)
@ -192,7 +180,6 @@ func TestHolder_HasData(t *testing.T) {
func TestHolder_DeleteIndex(t *testing.T) {
hldr := test.MustOpenHolder(t)
defer hldr.Close()
// Write bits to separate indexes.
hldr.SetBit("i0", "f", 100, 200)

2
http_translator_test.go
View file

@ -26,7 +26,7 @@ func TestTranslateStore_EntryReader(t *testing.T) {
hldr := test.Holder{Holder: primary.Server.Holder()}
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{Keys: true})
_, err := index.CreateField("f", pilosa.OptFieldTypeDefault())
_, err := index.CreateField("f")
if err != nil {
t.Fatal(err)
}

15
index.go
View file

@ -39,7 +39,6 @@ type Index struct {
fields map[string]*Field
broadcaster broadcaster
Schemator disco.Schemator
serializer Serializer
Stats stats.StatsClient
@ -84,7 +83,6 @@ func NewIndex(holder *Holder, path, name string) (*Index, error) {
holder: holder,
trackExistence: true,
Schemator: disco.NewInMemSchemator(),
serializer: NopSerializer,
translateStores: make(map[int]TranslateStore),
@ -715,7 +713,7 @@ func (i *Index) persistField(ctx context.Context, cfm *CreateFieldMessage) error
if b, err := i.serializer.Marshal(cfm); err != nil {
return errors.Wrap(err, "marshaling")
} else if err := i.Schemator.CreateField(ctx, cfm.Index, cfm.Field, b); errors.Cause(err) == disco.ErrFieldExists {
} else if err := i.holder.Schemator.CreateField(ctx, cfm.Index, cfm.Field, b); errors.Cause(err) == disco.ErrFieldExists {
return ErrFieldExists
} else if err != nil {
return errors.Wrapf(err, "writing field to disco: %s/%s", cfm.Index, cfm.Field)
@ -732,7 +730,7 @@ func (i *Index) persistUpdateField(ctx context.Context, cfm *CreateFieldMessage)
if b, err := i.serializer.Marshal(cfm); err != nil {
return errors.Wrap(err, "marshaling")
} else if err := i.Schemator.UpdateField(ctx, cfm.Index, cfm.Field, b); errors.Cause(err) == disco.ErrFieldDoesNotExist {
} else if err := i.holder.Schemator.UpdateField(ctx, cfm.Index, cfm.Field, b); errors.Cause(err) == disco.ErrFieldDoesNotExist {
return ErrFieldNotFound
} else if err != nil {
return errors.Wrapf(err, "writing field to disco: %s/%s", cfm.Index, cfm.Field)
@ -742,7 +740,7 @@ func (i *Index) persistUpdateField(ctx context.Context, cfm *CreateFieldMessage)
func (i *Index) UpdateField(ctx context.Context, name string, update FieldUpdate) (*CreateFieldMessage, error) {
// Get field from etcd
buf, err := i.Schemator.Field(ctx, i.name, name)
buf, err := i.holder.Schemator.Field(ctx, i.name, name)
if err != nil {
return nil, errors.Wrapf(err, "getting field '%s' from etcd", name)
}
@ -879,7 +877,6 @@ func (i *Index) newField(path, name string) (*Field, error) {
f.idx = i
f.Stats = i.Stats
f.broadcaster = i.broadcaster
f.schemator = i.Schemator
f.serializer = i.serializer
f.OpenTranslateStore = i.OpenTranslateStore
return f, nil
@ -902,7 +899,7 @@ func (i *Index) DeleteField(name string) error {
}
// Delete the field from etcd as the system of record.
if err := i.Schemator.DeleteField(context.TODO(), i.name, name); err != nil {
if err := i.holder.Schemator.DeleteField(context.TODO(), i.name, name); err != nil {
return errors.Wrapf(err, "deleting field from etcd: %s/%s", i.name, name)
}
@ -971,7 +968,3 @@ type importData struct {
func FormatQualifiedIndexName(index string) string {
return fmt.Sprintf("%s\x00", index)
}
func (i *Index) Txf() *TxFactory {
return i.holder.txf
}

11
index_internal_test.go
View file

@ -4,21 +4,12 @@ package pilosa
import (
"testing"
"github.com/featurebasedb/featurebase/v3/testhook"
)
// mustOpenIndex returns a new, opened index at a temporary path. Panic on error.
func mustOpenIndex(tb testing.TB, opt IndexOptions) *Index {
path, err := testhook.TempDirInDir(tb, *TempDir, "pilosa-index-")
if err != nil {
panic(err)
}
h := NewHolder(path, mustHolderConfig())
h := newTestHolder(tb)
index, err := h.CreateIndex("i", opt)
testhook.Cleanup(tb, func() {
h.Close()
})
if err != nil {
panic(err)

55
index_test.go
View file

@ -11,11 +11,10 @@ import (
"testing"
"time"
pilosa "github.com/featurebasedb/featurebase/v3"
"github.com/featurebasedb/featurebase/v3/disco"
"github.com/featurebasedb/featurebase/v3/pql"
"github.com/featurebasedb/featurebase/v3/test"
"github.com/featurebasedb/featurebase/v3/testhook"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/test"
"github.com/pkg/errors"
)
@ -24,10 +23,10 @@ const ShardWidth = pilosa.ShardWidth
// Ensure index can open and retrieve a field.
func TestIndex_CreateFieldIfNotExists(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
// Create field.
f, err := index.CreateFieldIfNotExists("f", pilosa.OptFieldTypeDefault())
f, err := index.CreateFieldIfNotExists("f")
if err != nil {
t.Fatal(err)
} else if f == nil {
@ -35,7 +34,7 @@ func TestIndex_CreateFieldIfNotExists(t *testing.T) {
}
// Retrieve existing field.
other, err := index.CreateFieldIfNotExists("f", pilosa.OptFieldTypeDefault())
other, err := index.CreateFieldIfNotExists("f")
if err != nil {
t.Fatal(err)
} else if f.Field != other.Field {
@ -51,7 +50,7 @@ func TestIndex_CreateField(t *testing.T) {
// Ensure time quantum can be set appropriately on a new field.
t.Run("TimeQuantum", func(t *testing.T) {
t.Run("Explicit", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
// Create field with explicit quantum.
f, err := index.CreateField("f", pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMDH"), "0"))
@ -66,7 +65,7 @@ func TestIndex_CreateField(t *testing.T) {
// Ensure time quantum can be set appropriately on a new field.
t.Run("TimeQuantumNoStandardView", func(t *testing.T) {
t.Run("Explicit", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
// Create field with explicit quantum with no standard view
f, err := index.CreateField("f", pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMDH"), "0", true))
@ -81,7 +80,7 @@ func TestIndex_CreateField(t *testing.T) {
// Ensure field can include range columns.
t.Run("BSIFields", func(t *testing.T) {
t.Run("Int", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
// Create field with schema and verify it exists.
if f, err := index.CreateField("f", pilosa.OptFieldTypeInt(-990, 1000)); err != nil {
@ -99,7 +98,7 @@ func TestIndex_CreateField(t *testing.T) {
})
t.Run("Timestamp", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
// Create field with schema and verify it exists.
if f, err := index.CreateField("f", pilosa.OptFieldTypeTimestamp(pilosa.DefaultEpoch, pilosa.TimeUnitSeconds)); err != nil {
@ -120,7 +119,7 @@ func TestIndex_CreateField(t *testing.T) {
// on field creation FieldOptions validation.
/*
t.Run("ErrRangeCacheAllowed", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
if _, err := index.CreateField("f", pilosa.FieldOptions{
CacheType: pilosa.CacheTypeRanked,
@ -130,7 +129,7 @@ func TestIndex_CreateField(t *testing.T) {
})
t.Run("BSIFieldsWithCacheTypeNone", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
if _, err := index.CreateField("f", pilosa.FieldOptions{
CacheType: pilosa.CacheTypeNone,
CacheSize: uint32(5),
@ -140,7 +139,7 @@ func TestIndex_CreateField(t *testing.T) {
})
t.Run("ErrFieldFieldsAllowed", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
if _, err := index.CreateField("f", pilosa.FieldOptions{
Fields: []*pilosa.Field{
@ -152,7 +151,7 @@ func TestIndex_CreateField(t *testing.T) {
})
t.Run("ErrFieldNameRequired", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
if _, err := index.CreateField("f", pilosa.FieldOptions{
Fields: []*pilosa.Field{
@ -164,7 +163,7 @@ func TestIndex_CreateField(t *testing.T) {
})
t.Run("ErrInvalidFieldType", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
if _, err := index.CreateField("f", pilosa.FieldOptions{
Fields: []*pilosa.Field{
@ -176,7 +175,7 @@ func TestIndex_CreateField(t *testing.T) {
})
t.Run("ErrInvalidBSIGroupRange", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
if _, err := index.CreateField("f", pilosa.FieldOptions{
Fields: []*pilosa.Field{
@ -192,7 +191,7 @@ func TestIndex_CreateField(t *testing.T) {
t.Run("WithKeys", func(t *testing.T) {
// Don't allow an int field to be created with keys=true
t.Run("IntField", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
_, err := index.CreateField("f", pilosa.OptFieldTypeInt(-1, 1), pilosa.OptFieldKeys())
if errors.Cause(err) != pilosa.ErrIntFieldWithKeys {
@ -202,7 +201,7 @@ func TestIndex_CreateField(t *testing.T) {
// Don't allow a decimal field to be created with keys=true
t.Run("DecimalField", func(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
_, err := index.CreateField("f", pilosa.OptFieldTypeDecimal(1, pql.NewDecimal(-1, 0), pql.NewDecimal(1, 0)), pilosa.OptFieldKeys())
if errors.Cause(err) != pilosa.ErrDecimalFieldWithKeys {
@ -214,10 +213,10 @@ func TestIndex_CreateField(t *testing.T) {
// Ensure index can delete a field.
func TestIndex_DeleteField(t *testing.T) {
index := test.MustOpenIndex(t)
_, index := test.MustOpenIndex(t)
// Create field.
if _, err := index.CreateFieldIfNotExists("f", pilosa.OptFieldTypeDefault()); err != nil {
if _, err := index.CreateFieldIfNotExists("f"); err != nil {
t.Fatal(err)
}
@ -237,11 +236,8 @@ func TestIndex_DeleteField(t *testing.T) {
// Ensure index can validate its name.
func TestIndex_InvalidName(t *testing.T) {
path, err := testhook.TempDir(t, "pilosa-index-")
if err != nil {
panic(err)
}
index, err := pilosa.NewIndex(pilosa.NewHolder(path, mustHolderConfig()), path, "ABC")
holder := test.NewHolder(t).Holder
index, err := pilosa.NewIndex(holder, holder.IndexPath("ABC"), "ABC")
if err == nil {
t.Fatalf("should have gotten an error on index name with caps")
}
@ -276,8 +272,7 @@ func TestIndex_RecreateFieldOnRestart(t *testing.T) {
// create field
fieldName := fmt.Sprintf("field_%d", rand.Uint64())
_, err = c.GetNode(0).API.CreateField(context.Background(), indexName, fieldName,
pilosa.OptFieldTypeDefault())
_, err = c.GetNode(0).API.CreateField(context.Background(), indexName, fieldName)
if err != nil {
t.Fatal(err)
}
@ -310,7 +305,7 @@ func TestIndex_RecreateFieldOnRestart(t *testing.T) {
errCh := make(chan error)
go func() {
_, err := c.GetNode(0).API.CreateField(context.Background(), indexName,
fieldName, pilosa.OptFieldTypeDefault())
fieldName)
errCh <- err
}()
select {

11
server.go
View file

@ -57,10 +57,9 @@ type Server struct { // nolint: maligned
serializer Serializer
// Distributed Consensus
disCo disco.DisCo
noder disco.Noder
sharder disco.Sharder
schemator disco.Schemator
disCo disco.DisCo
noder disco.Noder
sharder disco.Sharder
// External
systemInfo SystemInfo
@ -410,7 +409,7 @@ func OptServerDisCo(disCo disco.DisCo,
s.disCo = disCo
s.noder = noder
s.sharder = sharder
s.schemator = schemator
s.holderConfig.Schemator = schemator
return nil
}
}
@ -458,7 +457,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
disCo: disco.NopDisCo,
noder: disco.NewEmptyLocalNoder(),
sharder: disco.NopSharder,
schemator: disco.NopSchemator,
serializer: NopSerializer,
confirmDownRetries: defaultConfirmDownRetries,
@ -543,7 +541,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.cluster.confirmDownRetries = s.confirmDownRetries
s.cluster.confirmDownSleep = s.confirmDownSleep
s.holder.broadcaster = s
s.holder.schemator = s.schemator
s.holder.sharder = s.sharder
s.holder.serializer = s.serializer

2
server/grpc.go
View file

@ -752,7 +752,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errToStatusError(err)
}
qcx := index.Txf().NewQcx()
qcx := h.api.Holder().Txf().NewQcx()
defer qcx.Abort()
var fields []*pilosa.Field

24
server/handler_test.go
View file

@ -15,7 +15,6 @@ import (
"reflect"
"sort"
"strings"
"sync"
"testing"
"time"
@ -1337,33 +1336,18 @@ func TestHandler_Endpoints(t *testing.T) {
}
func TestCluster_TranslateStore(t *testing.T) {
cluster := test.MustNewCluster(t, 1)
cluster.Nodes[0] = test.NewCommandNode(t,
cluster := test.MustRunUnsharedCluster(t, 1, []server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})),
),
)
if err := cluster.Start(); err != nil {
t.Fatalf("starting node 0: %v", err)
}
defer cluster.GetIdleNode(0).Close() // nolint: errcheck
})
defer cluster.Close() // nolint: errcheck
test.Do(t, "POST", cluster.GetIdleNode(0).URL()+"/index/i0", "{\"options\": {\"keys\": true}}")
}
func TestClusterTranslator(t *testing.T) {
cluster := test.MustRunCluster(t, 3,
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})),
)},
cluster := test.MustRunUnsharedCluster(t, 3,
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),

2
server_internal_test.go
View file

@ -12,7 +12,7 @@ import (
func TestMonitorAntiEntropyZero(t *testing.T) {
td, err := testhook.TempDirInDir(t, *TempDir, "")
td, err := testhook.TempDir(t, "")
if err != nil {
t.Fatalf("getting temp dir: %v", err)
}

1
stats/stats_test.go
View file

@ -20,7 +20,6 @@ import (
// since the EXPVAR data is stored in a global we should run these in one test function
func TestMultiStatClient_Expvar(t *testing.T) {
hldr := test.MustOpenHolder(t)
defer hldr.Close()
c := stats.NewExpvarStatsClient()
ms := make(stats.MultiStatsClient, 1)

13
test/cluster.go
View file

@ -153,8 +153,10 @@ func (c *ShareableCluster) QueryGRPC(t testing.TB, index, query string) *proto.T
// sorted order. In other words, this method can only be used to retrieve a node
// when order doesn't matter. An example is if you need to do something like
// this:
// c.GetNode(0).Config.Cluster.ReplicaN = 2
// c.GetNode(1).Config.Cluster.ReplicaN = 2
//
// c.GetNode(0).Config.Cluster.ReplicaN = 2
// c.GetNode(1).Config.Cluster.ReplicaN = 2
//
// In this example, the test needs the replication factor to be set to 2 before
// starting; it's ok to reference each node by its index in the pre-sorted node
// list. It's also safe to use this method after `MustRunCluster()` if the
@ -423,7 +425,7 @@ type KeyID struct {
ID uint64
}
//ImportIDKey imports data into an unkeyed set field in a keyed index.
// ImportIDKey imports data into an unkeyed set field in a keyed index.
func (c *ShareableCluster) ImportIDKey(t testing.TB, index, field string, pairs []KeyID) {
t.Helper()
importRequest := &pilosa.ImportRequest{
@ -791,10 +793,7 @@ func prependOpts(opts [][]server.CommandOption, size int) [][]server.CommandOpti
// tweaks to initial startup delay, and storage config to disable fsync and
// specify a smaller RBF size.
func prependTestServerOpts(opts []server.CommandOption) []server.CommandOption {
cfg := pilosa.DefaultHolderConfig()
cfg.RBFConfig.FsyncEnabled = false
cfg.RBFConfig.MaxSize = (1 << 28)
cfg.RBFConfig.MaxWALSize = (1 << 28)
cfg := pilosa.TestHolderConfig()
defaultOpts := []server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(pilosa.OpenInMemTranslateStore),

35
test/holder.go
View file

@ -15,7 +15,8 @@ import (
// Holder is a test wrapper for pilosa.Holder.
type Holder struct {
*pilosa.Holder
tb testing.TB
tb testing.TB
closed bool
}
// NewHolder returns a new instance of Holder with a temporary path.
@ -25,10 +26,7 @@ func NewHolder(tb testing.TB) *Holder {
panic(err)
}
cfg := pilosa.DefaultHolderConfig()
cfg.StorageConfig.FsyncEnabled = false
cfg.RBFConfig.FsyncEnabled = false
cfg.RBFConfig.MaxSize = (1 << 28)
cfg := pilosa.TestHolderConfig()
h := &Holder{Holder: pilosa.NewHolder(path, cfg), tb: tb}
return h
@ -40,11 +38,22 @@ func MustOpenHolder(tb testing.TB) *Holder {
if err := h.Open(); err != nil {
tb.Fatalf("opening holder: %v", err)
}
tb.Cleanup(func() {
err := h.Close()
if err != nil {
tb.Fatalf("closing holder after test: %v", err)
}
})
return h
}
// Close closes the holder. The data should be removed by the
// Close closes the holder. The data should be removed by the cleanup
// registered when we created the initial tempdir.
func (h *Holder) Close() error {
if h.closed {
h.tb.Fatal("double-closed holder")
}
h.closed = true
return h.Holder.Close()
}
@ -70,7 +79,7 @@ func (h *Holder) Row(index, field string, rowID uint64) *pilosa.Row {
if err != nil {
panic(err)
}
qcx := idx.Txf().NewQcx()
qcx := h.Txf().NewQcx()
defer qcx.Abort()
row, err := f.Row(qcx, rowID)
@ -93,7 +102,7 @@ func (h *Holder) ReadRow(index, field string, rowID uint64) *pilosa.Row {
if f == nil {
h.tb.Fatalf("read row from field %q/%q: field not found", index, field)
}
qcx := idx.Txf().NewQcx()
qcx := h.Txf().NewQcx()
defer qcx.Abort()
row, err := f.Row(qcx, rowID)
@ -112,7 +121,7 @@ func (h *Holder) RowTime(index, field string, rowID uint64, t time.Time, quantum
if err != nil {
panic(err)
}
qcx := idx.Txf().NewQcx()
qcx := h.Txf().NewQcx()
defer qcx.Abort()
row, err := f.RowTime(qcx, rowID, t, quantum)
@ -138,7 +147,7 @@ func (h *Holder) SetBitTime(index, field string, rowID, columnID uint64, t *time
panic(err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := h.Txf().NewWritableQcx()
defer qcx.Abort()
_, err = f.SetBit(qcx, rowID, columnID, t)
@ -159,7 +168,7 @@ func (h *Holder) ClearBit(index, field string, rowID, columnID uint64) {
panic(err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := h.Txf().NewWritableQcx()
defer qcx.Abort()
_, err = f.ClearBit(qcx, rowID, columnID)
@ -188,7 +197,7 @@ func (h *Holder) SetValue(index, field string, columnID uint64, value int64) *In
panic(err)
}
qcx := idx.Txf().NewWritableQcx()
qcx := h.Txf().NewWritableQcx()
defer qcx.Abort()
_, err = f.SetValue(qcx, columnID, value)
if err != nil {
@ -210,7 +219,7 @@ func (h *Holder) Value(index, field string, columnID uint64) (int64, bool) {
panic(err)
}
qcx := idx.Txf().NewQcx()
qcx := h.Txf().NewQcx()
defer qcx.Abort()
val, exists, err := f.Value(qcx, columnID)

30
test/index.go
View file

@ -6,8 +6,8 @@ import (
"context"
"testing"
pilosa "github.com/featurebasedb/featurebase/v3"
"github.com/featurebasedb/featurebase/v3/testhook"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/testhook"
)
// Index represents a test wrapper for pilosa.Index.
@ -15,17 +15,9 @@ type Index struct {
*pilosa.Index
}
// newIndex returns a new instance of Index.
func newIndex(tb testing.TB) *Index {
path, err := testhook.TempDir(tb, "pilosa-index-")
if err != nil {
panic(err)
}
cfg := pilosa.DefaultHolderConfig()
cfg.StorageConfig.FsyncEnabled = false
cfg.RBFConfig.FsyncEnabled = false
cfg.RBFConfig.MaxSize = (1 << 28)
h := pilosa.NewHolder(path, cfg)
// newIndex returns a new instance of Index, and the parent holder.
func newIndex(tb testing.TB) (*Holder, *Index) {
h := NewHolder(tb)
testhook.Cleanup(tb, func() {
h.Close()
})
@ -33,13 +25,13 @@ func newIndex(tb testing.TB) *Index {
if err != nil {
panic(err)
}
return &Index{Index: index}
return h, &Index{Index: index}
}
// MustOpenIndex returns a new, opened index at a temporary path. Panic on error.
func MustOpenIndex(tb testing.TB) *Index {
index := newIndex(tb)
return index
// MustOpenIndex returns a new, opened index at a temporary path, or
// fails the test. It also returns the holder containing the index.
func MustOpenIndex(tb testing.TB) (*Holder, *Index) {
return newIndex(tb)
}
// Close closes the index and removes the underlying data.
@ -52,7 +44,7 @@ func (i *Index) Reopen() error {
if err := i.Index.Close(); err != nil {
return err
}
schema, err := i.Schemator.Schema(context.Background())
schema, err := i.Holder().Schemator.Schema(context.Background())
if err != nil {
return err
}

6
tx_internal_test.go
View file

@ -57,7 +57,7 @@ func requireCountRangeSampleData(tb testing.TB) (*fragment, Tx) {
countRangeSampleData = asBytes.Bytes()
tb.Logf("creating bitmap: %d containers, %d bytes of data", countRangeMaxN, n)
})
f, idx, tx := mustOpenFragment(tb, "i", "f", viewStandard, 0, "")
f, idx, tx := mustOpenFragment(tb)
// Properly close this transaction, but not the next one we create that the
// caller will be responsible for. The deferred callback will
// be a nop if the Commit happened.
@ -94,7 +94,7 @@ func TestTx_CountRange(t *testing.T) {
}
// Every other bit gets set, for a total of i bits in container
// i, so they're all in the first (i*2) bits of the container.
got, err := tx.CountRange("i", "f", viewStandard, 0, uint64(j)<<16, (uint64(i)<<16)+(i*2))
got, err := tx.CountRange("i", "f", "v", 0, uint64(j)<<16, (uint64(i)<<16)+(i*2))
if err != nil {
t.Fatalf("counting range: %v", err)
}
@ -120,7 +120,7 @@ func BenchmarkTx_CountRange(b *testing.B) {
expected -= (j * 7) + 21
j += 7
}
got, err := tx.CountRange("i", "f", viewStandard, 0, uint64(j)<<16, uint64(i)<<16)
got, err := tx.CountRange("i", "f", "v", 0, uint64(j)<<16, uint64(i)<<16)
if err != nil {
b.Fatalf("counting range: %v", err)
}

23
view.go
View file

@ -280,15 +280,6 @@ func (v *view) flushCaches() {
}
}
// flags returns a set of flags for the underlying fragments.
func (v *view) flags() byte {
var flag byte
if v.fieldType == FieldTypeInt || v.fieldType == FieldTypeDecimal || v.fieldType == FieldTypeTimestamp {
flag |= roaringFlagBSIv2
}
return flag
}
// availableShards returns a bitmap of shards which contain data.
func (v *view) availableShards() *roaring.Bitmap {
// A read lock prevents anything with the write lock from being
@ -402,19 +393,7 @@ func (v *view) notifyIfNewShard(shard uint64) {
}
func (v *view) newFragment(shard uint64) *fragment {
fld := v.fld
spec := fragSpec{
index: v.idx,
field: fld,
view: v,
}
if fld == nil {
// The backup plan.
// For tests that do incomplete setup, like making
// a view without a field.
spec.fieldstr = v.field
}
frag := newFragment(v.holder, spec, shard, v.flags())
frag := newFragment(v.holder, v.idx, v.fld, v, shard)
frag.CacheType = v.cacheType
frag.CacheSize = v.cacheSize
frag.stats = v.stats

40
view_internal_test.go
View file

@ -6,50 +6,21 @@ import (
"testing"
"time"
"github.com/featurebasedb/featurebase/v3/testhook"
. "github.com/featurebasedb/featurebase/v3/vprint" // nolint:staticcheck
"golang.org/x/sync/errgroup"
)
// mustOpenView returns a new instance of View with a temporary path.
func mustOpenView(tb testing.TB, index, field, name string) *view {
path, err := testhook.TempDirInDir(tb, *TempDir, "pilosa-view-")
if err != nil {
PanicOn(err)
}
fo := FieldOptions{
CacheType: DefaultCacheType,
CacheSize: DefaultCacheSize,
}
h := NewHolder(path, mustHolderConfig())
// h needs an *Index so we can call h.Index() and get Index.Txf, in TestView_DeleteFragment
cim := &CreateIndexMessage{
Index: index,
CreatedAt: 0,
Meta: IndexOptions{},
}
idx, err := h.createIndex(cim, false)
testhook.Cleanup(tb, func() {
h.Close()
})
PanicOn(err)
v := newView(h, path, index, field, name, fo)
v.idx = idx
func mustOpenView(tb testing.TB) *view {
_, _, _, v := newTestView(tb)
if err := v.openEmpty(); err != nil {
PanicOn(err)
tb.Fatalf("opening empty test view: %v", err)
}
return v
}
// Ensure view can open and retrieve a fragment.
func TestView_DeleteFragment(t *testing.T) {
v := mustOpenView(t, "i", "f", "v")
defer v.close()
v := mustOpenView(t)
shard := uint64(9)
@ -83,8 +54,7 @@ func TestView_DeleteFragment(t *testing.T) {
// if the broadcast operation takes a bit of time.
func TestView_CreateFragmentRace(t *testing.T) {
var creates errgroup.Group
v := mustOpenView(t, "i", "f", "v")
defer v.close()
v := mustOpenView(t)
// Use a broadcaster which intentionally fails.
v.broadcaster = delayBroadcaster{delay: 10 * time.Millisecond}