diff --git a/api.go b/api.go index 1e1bb5a7f..dacd90ef9 100644 --- a/api.go +++ b/api.go @@ -2737,7 +2737,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 } diff --git a/dbshard_internal_test.go b/dbshard_internal_test.go index dcb9c912e..208b8fe0d 100644 --- a/dbshard_internal_test.go +++ b/dbshard_internal_test.go @@ -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() diff --git a/executor.go b/executor.go index 49c1f38ff..4f9035834 100644 --- a/executor.go +++ b/executor.go @@ -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 diff --git a/executor_internal_test.go b/executor_internal_test.go index 2f3cb45cd..6add1b1f9 100644 --- a/executor_internal_test.go +++ b/executor_internal_test.go @@ -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) diff --git a/executor_test.go b/executor_test.go index 85b5ec4d6..fb18ce87e 100644 --- a/executor_test.go +++ b/executor_test.go @@ -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) } diff --git a/field.go b/field.go index b0e23b76c..9f76c0add 100644 --- a/field.go +++ b/field.go @@ -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. diff --git a/field_internal_test.go b/field_internal_test.go index 48b5bbb78..1193d9ef1 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -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< /tmp/holder-dir199550572/i/f/views/standard - splt := strings.Split(path, sep+"fragments"+sep) - return fragSpec{ - index: &Index{path: splt[0], name: index}, - field: &Field{name: field}, - view: &view{path: splt[0], name: view0}, - } -} - func BenchmarkFragment_Import(b *testing.B) { b.StopTimer() maxX := ShardWidth * 5 * 2 @@ -2545,7 +2531,7 @@ func BenchmarkFragment_Import(b *testing.B) { // since bulkImport modifies the input slices, we make new copies for each round copy(rowsUse, rows) copy(colsUse, cols) - f, idx, tx := mustOpenFragment(b, "i", "f", viewStandard, 0, "") + f, idx, tx := mustOpenFragment(b) _ = idx b.StartTimer() if err := f.bulkImport(tx, rowsUse, colsUse, options); err != nil { @@ -2572,7 +2558,7 @@ func BenchmarkImportRoaring(b *testing.B) { b.Run(fmt.Sprintf("Rows%dCache_%s", numRows, cacheType), func(b *testing.B) { b.StopTimer() for i := 0; i < b.N; i++ { - f, _, tx := mustOpenFragment(b, "i", fmt.Sprintf("r%dc%s", numRows, cacheType), viewStandard, 0, cacheType) + f, _, tx := mustOpenFragment(b, OptFieldTypeSet(cacheType, 0)) b.StartTimer() err := f.importRoaringT(tx, data, false) @@ -2608,7 +2594,7 @@ func BenchmarkImportRoaringConcurrent(b *testing.B) { txs := make([]Tx, concurrency) for i := 0; i < b.N; i++ { for j := 0; j < concurrency; j++ { - frags[j], _, txs[j] = mustOpenFragment(b, "i", "f", viewStandard, uint64(j), cacheType) + frags[j], _, txs[j] = mustOpenFragment(b, OptFieldTypeSet(cacheType, 0)) } eg := errgroup.Group{} b.StartTimer() @@ -2646,7 +2632,7 @@ func BenchmarkImportStandard(b *testing.B) { for i := 0; i < b.N; i++ { copy(rowIDs, rowIDsOrig) copy(columnIDs, columnIDsOrig) - f, idx, tx := mustOpenFragment(b, "i", fmt.Sprintf("r%dc%s", numRows, cacheType), viewStandard, 0, cacheType) + f, idx, tx := mustOpenFragment(b, OptFieldTypeSet(cacheType, 0)) _ = idx b.StartTimer() err := f.bulkImport(tx, rowIDs, columnIDs, &ImportOptions{}) @@ -2675,7 +2661,7 @@ func BenchmarkImportRoaringUpdate(b *testing.B) { b.Run(name, func(b *testing.B) { b.StopTimer() for i := 0; i < b.N; i++ { - f, idx, tx := mustOpenFragment(b, "i", fmt.Sprintf("r%dc%dcache_%s", numRows, numCols, cacheType), viewStandard, 0, cacheType) + f, idx, tx := mustOpenFragment(b, OptFieldTypeSet(cacheType, 0)) _ = idx itr, err := roaring.NewRoaringIterator(data) @@ -2728,7 +2714,7 @@ func BenchmarkUpdatePathological(b *testing.B) { b.ResetTimer() for i := 0; i < b.N; i++ { b.StopTimer() - f, idx, tx := mustOpenFragment(b, "i", "f", viewStandard, 0, DefaultCacheType) + f, idx, tx := mustOpenFragment(b, OptFieldTypeSet(DefaultCacheType, 0)) _ = idx err := f.importRoaringT(tx, exists, false) @@ -2749,7 +2735,7 @@ var bigFrag string func initBigFrag(tb testing.TB) { if bigFrag == "" { - f, _, tx := mustOpenFragment(tb, "i", "f", viewStandard, 0, DefaultCacheType) + f, _, tx := mustOpenFragment(tb, OptFieldTypeSet(DefaultCacheType, 0)) for i := int64(0); i < 10; i++ { // 10 million rows, 1 bit per column, random seeded by i data := getZipfRowsSliceRoaring(10000000, i, 0, ShardWidth) @@ -2778,7 +2764,7 @@ func BenchmarkImportIntoLargeFragment(b *testing.B) { if err != nil { b.Fatalf("opening frag file: %v", err) } - fi, err := testhook.TempFileInDir(b, *TempDir, "") + fi, err := testhook.TempFile(b, "") if err != nil { b.Fatalf("getting temp file: %v", err) } @@ -2789,19 +2775,15 @@ func BenchmarkImportIntoLargeFragment(b *testing.B) { origF.Close() fi.Close() - h := NewHolder(fi.Name(), mustHolderConfig()) - PanicOn(h.Open()) - idx, err := h.CreateIndex("i", IndexOptions{}) - PanicOn(err) + h, idx, _, _, f := newTestFragment(b) - f := newFragment(h, makeTestFragSpec(fi.Name(), "i", "f", viewStandard), 0, 0) err = f.Open() if err != nil { b.Fatalf("opening fragment: %v", err) } // Obtain transaction. - tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard}) + tx := h.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard}) defer tx.Rollback() copy(rows, rowsOrig) @@ -2814,7 +2796,6 @@ func BenchmarkImportIntoLargeFragment(b *testing.B) { } PanicOn(tx.Commit()) f.Clean(b) - h.Close() } } @@ -2822,13 +2803,12 @@ func BenchmarkImportRoaringIntoLargeFragment(b *testing.B) { b.StopTimer() initBigFrag(b) updata := getUpdataRoaring(10000000, 11000, 0) - th := newTestHolder(b) for i := 0; i < b.N; i++ { origF, err := os.Open(bigFrag) if err != nil { b.Fatalf("opening frag file: %v", err) } - fi, err := testhook.TempFileInDir(b, *TempDir, "") + fi, err := testhook.TempFile(b, "") if err != nil { b.Fatalf("getting temp file: %v", err) } @@ -2839,15 +2819,11 @@ func BenchmarkImportRoaringIntoLargeFragment(b *testing.B) { origF.Close() fi.Close() - // want to do this, but no path argument. - //nf, idx, tx := mustOpenFragmentFlags(index, field, view string, shard uint64, cacheType string, flags byte) + h, idx, _, _, f := newTestFragment(b) - idx := fragTestMustOpenIndex("i", th, IndexOptions{}) - // XXX TODO: newFragment is using the wrong path here, we should fix that someday. - f := newFragment(th, makeTestFragSpec(fi.Name(), "i", "f", viewStandard), 0, 0) defer f.Clean(b) - tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard}) + tx := h.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard}) defer tx.Rollback() err = f.Open() @@ -2864,7 +2840,7 @@ func BenchmarkImportRoaringIntoLargeFragment(b *testing.B) { } func TestGetZipfRowsSliceRoaring(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, DefaultCacheType) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(DefaultCacheType, 0)) _ = idx data := getZipfRowsSliceRoaring(10, 1, 0, ShardWidth) @@ -2887,7 +2863,7 @@ func TestGetZipfRowsSliceRoaring(t *testing.T) { } func prepareSampleRowData(b *testing.B, bits int, rows uint64, width uint64) (*fragment, *Index, Tx) { - f, idx, tx := mustOpenFragment(b, "i", "f", viewStandard, 0, "none") + f, idx, tx := mustOpenFragment(b, OptFieldTypeSet("none", 0)) for i := 0; i < bits; i++ { data := getUniformRowsSliceRoaring(rows, int64(rows)+int64(i), 0, width) err := f.importRoaringT(tx, data, false) @@ -3075,44 +3051,6 @@ func getZipfRowsSliceStandard(numRows uint64, seed int64, startCol, endCol uint6 return rowIDs, columnIDs } -func BenchmarkFileWrite(b *testing.B) { - for _, numRows := range rowCases { - data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth) - b.Run(fmt.Sprintf("Rows%d", numRows), func(b *testing.B) { - b.StopTimer() - for i := 0; i < b.N; i++ { - // DO NOT CONVERT THIS ONE TO USE TESTHOOK. - // We're deleting these files as we go because - // otherwise the benchmark could fill up - // $TMPDIR before it finishes running. - f, err := os.CreateTemp(*TempDir, "") - if err != nil { - b.Fatalf("getting temp file: %v", err) - } - b.StartTimer() - _, err = f.Write(data) - if err != nil { - os.Remove(f.Name()) - b.Fatal(err) - } - err = f.Sync() - if err != nil { - os.Remove(f.Name()) - b.Fatal(err) - } - err = f.Close() - if err != nil { - os.Remove(f.Name()) - b.Fatal(err) - } - b.StopTimer() - os.Remove(f.Name()) - } - }) - } - -} - // Clean used to delete fragments, but doesn't anymore -- deleting is // handled by the testhook.TempDir when appropriate. // TODO(jaffee): this can likely go away entirely... it was doing snapshot/source/generation stuff that it no longer needs to. @@ -3129,19 +3067,13 @@ func (f *fragment) importRoaringT(tx Tx, data []byte, clear bool) error { return f.importRoaring(context.Background(), tx, data, clear) } -// mustOpenFragment returns a new instance of Fragment with a temporary path. -func mustOpenFragment(tb testing.TB, index, field, view string, shard uint64, cacheType string) (*fragment, *Index, Tx) { - return mustOpenFragmentFlags(tb, index, field, view, shard, cacheType, 0) -} - -func mustOpenBSIFragment(tb testing.TB, index, field, view string, shard uint64) (*fragment, *Index, Tx) { - return mustOpenFragmentFlags(tb, index, field, view, shard, "", 1) -} - func newTestHolder(tb testing.TB) *Holder { - path, _ := testhook.TempDirInDir(tb, *TempDir, "holder-dir") - h := NewHolder(path, mustHolderConfig()) - PanicOn(h.Open()) + path := tb.TempDir() + h := NewHolder(path, TestHolderConfig()) + err := h.Open() + if err != nil { + tb.Fatalf("opening test holder: %v", err) + } testhook.Cleanup(tb, func() { h.Close() }) @@ -3149,70 +3081,67 @@ func newTestHolder(tb testing.TB) *Holder { return h } -// fragTestMustOpenIndex returns a new, opened index at a temporary path. Panic on error. -func fragTestMustOpenIndex(index string, holder *Holder, opt IndexOptions) *Index { - cim := &CreateIndexMessage{ - Index: index, - CreatedAt: 0, - Meta: opt, +// newTestField creates a field with the specified field options +func newTestField(tb testing.TB, fieldOpts ...FieldOption) (*Holder, *Index, *Field) { + if len(fieldOpts) == 0 { + fieldOpts = []FieldOption{OptFieldTypeDefault()} } - - holder.mu.Lock() - idx, err := holder.createIndex(cim, false) - holder.mu.Unlock() - PanicOn(err) - - idx.keys = opt.Keys - idx.trackExistence = opt.TrackExistence - - if err := idx.Open(); err != nil { - PanicOn(err) + h := newTestHolder(tb) + idx, err := h.CreateIndex("i", IndexOptions{}) + if err != nil { + tb.Fatalf("creating test index: %v", err) } - return idx + fld, err := idx.CreateField("f", fieldOpts...) + if err != nil { + tb.Fatalf("creating test field: %v", err) + } + return h, idx, fld +} + +// newTestView creates a view with the specified options. +func newTestView(tb testing.TB, fieldOpts ...FieldOption) (*Holder, *Index, *Field, *view) { + h, idx, fld := newTestField(tb, fieldOpts...) + v, _, err := fld.createViewIfNotExistsBase(&CreateViewMessage{Index: "i", Field: "f", View: "v"}) + if err != nil { + tb.Fatalf("creating test view: %v", err) + } + return h, idx, fld, v +} + +// newTestFragment makes the default /i/f/v/0 fragment, and returns the +// things. the test holder will be deleted automatically in test cleanup. +func newTestFragment(tb testing.TB, fieldOpts ...FieldOption) (*Holder, *Index, *Field, *view, *fragment) { + h, idx, fld, v := newTestView(tb, fieldOpts...) + f := v.newFragment(0) + return h, idx, fld, v, f } // mustOpenFragment returns a new instance of Fragment with a temporary path. -func mustOpenFragmentFlags(tb testing.TB, index, field, view string, shard uint64, cacheType string, flags byte) (*fragment, *Index, Tx) { - if cacheType == "" { - cacheType = DefaultCacheType +func mustOpenFragment(tb testing.TB, fieldOpts ...FieldOption) (*fragment, *Index, Tx) { + th, idx, fld, v, f := newTestFragment(tb, fieldOpts...) + fragDir := filepath.Join(idx.path, fld.name, "views", v.name, "fragments") + err := os.MkdirAll(fragDir, 0700) + if err != nil { + tb.Fatalf("creating fragment directory: %v", err) } - th := newTestHolder(tb) - idx := fragTestMustOpenIndex(index, th, IndexOptions{}) - - fragDir := fmt.Sprintf("%v/%v/views/%v/fragments/", idx.path, field, view) - PanicOn(os.MkdirAll(fragDir, 0750)) - fragPath := fragDir + fmt.Sprintf("%v", shard) - f := newFragment(th, makeTestFragSpec(fragPath, index, field, view), shard, flags) - - tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: shard}) + tx := th.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: 0}) testhook.Cleanup(tb, func() { tx.Rollback() - PanicOn(idx.holder.txf.CloseIndex(idx)) + + if err := th.txf.CloseIndex(idx); err != nil { + tb.Fatalf("closing index after test: %v", err) + } }) - f.CacheType = cacheType + f.CacheType = fld.options.CacheType if err := f.Open(); err != nil { - PanicOn(err) + tb.Fatalf("opening fragment: %v", err) } return f, idx, tx } -// mustOpenMutexFragment returns a new instance of Fragment for a mutex field. -func mustOpenMutexFragment(tb testing.TB, index, field, view string, shard uint64, cacheType string) (*fragment, *Index, Tx) { - frag, idx, tx := mustOpenFragment(tb, index, field, view, shard, cacheType) - frag.mutexVector = newRowsVector(frag) - return frag, idx, tx -} - -// mustOpenBoolFragment returns a new instance of Fragment for a bool field. -func mustOpenBoolFragment(tb testing.TB, index, field, view string, shard uint64, cacheType string) (*fragment, *Index, Tx) { - frag, idx, tx := mustOpenFragment(tb, index, field, view, shard, cacheType) - frag.mutexVector = newBoolVector(frag) - return frag, idx, tx -} - // Reopen closes the fragment and reopens it as a new instance. func (f *fragment) Reopen() error { if err := f.Close(); err != nil { @@ -3245,7 +3174,7 @@ func addToBitmap(bm *roaring.Bitmap, rowID uint64, columnIDs ...uint64) { // Test Various methods of retrieving RowIDs func TestFragment_RowsIteration(t *testing.T) { t.Run("firstContainer", func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") + f, idx, tx := mustOpenFragment(t) _ = idx defer f.Clean(t) @@ -3277,7 +3206,7 @@ func TestFragment_RowsIteration(t *testing.T) { }) t.Run("secondRow", func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") + f, idx, tx := mustOpenFragment(t) _ = idx defer f.Clean(t) @@ -3311,7 +3240,7 @@ func TestFragment_RowsIteration(t *testing.T) { }) t.Run("combinations", func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") + f, idx, tx := mustOpenFragment(t) _ = idx defer f.Clean(t) @@ -3364,7 +3293,7 @@ func TestFragment_RoaringImport(t *testing.T) { for i, test := range tests { t.Run(fmt.Sprintf("importroaring%d", i), func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") + f, idx, tx := mustOpenFragment(t) _ = idx defer f.Clean(t) @@ -3413,7 +3342,7 @@ func TestFragment_RoaringImportTopN(t *testing.T) { for i, test := range tests { t.Run(fmt.Sprintf("importroaring%d", i), func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, CacheTypeRanked) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) _ = idx defer f.Clean(t) @@ -3552,7 +3481,7 @@ func calcExpected(inputs ...[]uint64) [][]uint64 { func TestFragmentRowIterator(t *testing.T) { t.Run("basic", func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeRanked) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) _ = idx defer f.Clean(t) @@ -3597,7 +3526,7 @@ func TestFragmentRowIterator(t *testing.T) { }) t.Run("skipped rows", func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeRanked) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) _ = idx defer f.Clean(t) @@ -3641,7 +3570,7 @@ func TestFragmentRowIterator(t *testing.T) { }) t.Run("basic wrapped", func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeRanked) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) _ = idx defer f.Clean(t) @@ -3674,7 +3603,7 @@ func TestFragmentRowIterator(t *testing.T) { }) t.Run("skipped rows wrapped", func(t *testing.T) { - f, _, tx := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeRanked) + f, _, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) defer f.Clean(t) f.mustSetBits(tx, 1, 0) @@ -3709,7 +3638,7 @@ func TestFragmentRowIterator(t *testing.T) { // same, with commits func TestFragmentRowIterator_WithTxCommit(t *testing.T) { t.Run("basic", func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeRanked) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) _ = idx defer f.Clean(t) @@ -3758,7 +3687,7 @@ func TestFragmentRowIterator_WithTxCommit(t *testing.T) { }) t.Run("skipped rows", func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeRanked) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) _ = idx defer f.Clean(t) @@ -3806,7 +3735,7 @@ func TestFragmentRowIterator_WithTxCommit(t *testing.T) { }) t.Run("basic wrapped", func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeRanked) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) _ = idx defer f.Clean(t) @@ -3843,7 +3772,7 @@ func TestFragmentRowIterator_WithTxCommit(t *testing.T) { }) t.Run("skipped rows wrapped", func(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeRanked) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) _ = idx defer f.Clean(t) @@ -3883,7 +3812,7 @@ func TestFragmentRowIterator_WithTxCommit(t *testing.T) { } func TestFragmentPositionsForValue(t *testing.T) { - f, _, _ := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeNone) + f, _, _ := mustOpenFragment(t, OptFieldTypeSet(CacheTypeNone, 0)) defer f.Clean(t) tests := []struct { @@ -3965,7 +3894,7 @@ func TestFragmentPositionsForValue(t *testing.T) { } func TestIntLTRegression(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeNone) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeNone, 0)) _ = idx defer f.Clean(t) @@ -3996,7 +3925,7 @@ func sliceEq(x, y []uint64) bool { func TestFragmentBSIUnsigned(t *testing.T) { shard := uint64(0) - f, idx, tx := mustOpenFragment(t, "i", "f", "v", shard, CacheTypeNone) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeNone, 0)) defer f.Clean(t) // Number of bits to test. @@ -4178,7 +4107,7 @@ func TestFragmentBSIUnsigned(t *testing.T) { // same, WithTxCommit version func TestFragmentBSIUnsigned_WithTxCommit(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeNone) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeNone, 0)) _ = idx defer f.Clean(t) @@ -4340,7 +4269,7 @@ func TestFragmentBSIUnsigned_WithTxCommit(t *testing.T) { } func TestFragmentBSISigned(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeNone) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeNone, 0)) _ = idx defer f.Clean(t) @@ -4503,7 +4432,7 @@ func TestFragmentBSISigned(t *testing.T) { } func TestImportValueConcurrent(t *testing.T) { - f, idx, tx := mustOpenBSIFragment(t, "i", "f", viewBSIGroupPrefix+"foo", 0) + f, idx, tx := mustOpenFragment(t) defer f.Clean(t) // we will be making a new Tx each time, so we can rollback the default provided one. tx.Rollback() @@ -4548,7 +4477,7 @@ func TestImportMultipleValues(t *testing.T) { for i, test := range tests { t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { - f, _, tx := mustOpenBSIFragment(t, "i", "f", viewBSIGroupPrefix+"foo", 0) + f, _, tx := mustOpenFragment(t) defer f.Clean(t) err := f.importValue(tx, test.cols, test.vals, test.depth, false) @@ -4602,7 +4531,7 @@ func TestImportValueRowCache(t *testing.T) { for i, test := range tests { t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { - f, _, tx := mustOpenBSIFragment(t, "i", "f", viewBSIGroupPrefix+"foo", 0) + f, _, tx := mustOpenFragment(t) defer f.Clean(t) // First import (tc1) @@ -4634,7 +4563,7 @@ func TestImportValueRowCache(t *testing.T) { // do we see races/corruption around concurrent read/write. // especially on writes to the row cache. func TestFragmentConcurrentReadWrite(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, CacheTypeRanked) + f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) defer f.Clean(t) tx.Rollback() @@ -4670,7 +4599,7 @@ func TestFragmentConcurrentReadWrite(t *testing.T) { } func TestFragment_Bug_Q2DoubleDelete(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") + f, idx, tx := mustOpenFragment(t) _ = idx // byShardWidth is a map of the same roaring (fragment) data generated // with different shard widths. @@ -4922,7 +4851,7 @@ func TestImportMutexSampleData(t *testing.T) { seen := make(map[uint64]struct{}) t.Run(data.name, func(t *testing.T) { batchSize := 16384 - f, _, tx := mustOpenMutexFragment(t, "i", "f", viewStandard, 0, "") + f, _, tx := mustOpenFragment(t, OptFieldTypeMutex(DefaultCacheType, DefaultCacheSize)) defer f.Clean(t) // Set import. var err error @@ -5000,7 +4929,7 @@ func BenchmarkImportMutexSampleData(b *testing.B) { } } benchmarkFragmentImports := func(b *testing.B) { - frag, idx, tx = mustOpenMutexFragment(b, "i", "f", viewStandard, 0, cache) + frag, idx, tx = mustOpenFragment(b, OptFieldTypeMutex(cache, DefaultCacheSize)) defer frag.Clean(b) for i := range data.colIDs { b.Run(fmt.Sprintf("write-%d", i), func(b *testing.B) { @@ -5288,7 +5217,7 @@ func TestSliceDifference(t *testing.T) { } func TestImportRoaringSingleValued(t *testing.T) { - f, _, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") + f, _, tx := mustOpenFragment(t) defer f.Clean(t) clear := roaring.NewBitmap(0, 1, ShardWidth-1) @@ -5324,7 +5253,7 @@ func TestImportRoaringSingleValued(t *testing.T) { t.Fatalf("importing: %v", err) } - result, err := tx.RoaringBitmap("i", "f", viewStandard, 0) + result, err := tx.RoaringBitmap("i", "f", "v", 0) if err != nil { t.Fatalf("getting bitmap: %v", err) } diff --git a/holder.go b/holder.go index c39509c6d..c9993b081 100644 --- a/holder.go +++ b/holder.go @@ -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. diff --git a/holder_internal_test.go b/holder_internal_test.go index 08634ed90..e9faa3a12 100644 --- a/holder_internal_test.go +++ b/holder_internal_test.go @@ -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) diff --git a/holder_test.go b/holder_test.go index 1e962d739..f6cc800f6 100644 --- a/holder_test.go +++ b/holder_test.go @@ -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) diff --git a/http_translator_test.go b/http_translator_test.go index 5f6bfcc9e..739bedef9 100644 --- a/http_translator_test.go +++ b/http_translator_test.go @@ -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) } diff --git a/index.go b/index.go index 4ab793c24..557983ec0 100644 --- a/index.go +++ b/index.go @@ -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 -} diff --git a/index_internal_test.go b/index_internal_test.go index 44afe8b94..e275069cf 100644 --- a/index_internal_test.go +++ b/index_internal_test.go @@ -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) diff --git a/index_test.go b/index_test.go index a60b948c5..fe6b9ed0d 100644 --- a/index_test.go +++ b/index_test.go @@ -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 { diff --git a/server.go b/server.go index 7213a8f8b..cbfaa3fa8 100644 --- a/server.go +++ b/server.go @@ -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 diff --git a/server/grpc.go b/server/grpc.go index 7cb714c34..350ac8c73 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -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 diff --git a/server/handler_test.go b/server/handler_test.go index 4f193f66e..fa77b0146 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -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), diff --git a/server_internal_test.go b/server_internal_test.go index 1b51fbaa7..eecea6ca0 100644 --- a/server_internal_test.go +++ b/server_internal_test.go @@ -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) } diff --git a/stats/stats_test.go b/stats/stats_test.go index ebc62d77c..ae0824be6 100644 --- a/stats/stats_test.go +++ b/stats/stats_test.go @@ -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) diff --git a/test/cluster.go b/test/cluster.go index 557906741..a1294cd82 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -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), diff --git a/test/holder.go b/test/holder.go index 409a64fb9..a8219abf3 100644 --- a/test/holder.go +++ b/test/holder.go @@ -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) diff --git a/test/index.go b/test/index.go index b1872bcdd..f8cd34756 100644 --- a/test/index.go +++ b/test/index.go @@ -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 } diff --git a/tx_internal_test.go b/tx_internal_test.go index 05efdacae..aa4992670 100644 --- a/tx_internal_test.go +++ b/tx_internal_test.go @@ -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) } diff --git a/view.go b/view.go index 5c1b08daf..df9b269e2 100644 --- a/view.go +++ b/view.go @@ -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 diff --git a/view_internal_test.go b/view_internal_test.go index 82bdda503..40cec1e19 100644 --- a/view_internal_test.go +++ b/view_internal_test.go @@ -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}