From ea69b0637d89a9ff50248774be8517e7d94165ad Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 21 Sep 2022 14:09:21 -0500 Subject: [PATCH] significant refactor of test setup and teardown 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. --- api.go | 2 +- dbshard_internal_test.go | 28 +-- executor.go | 23 ++- executor_internal_test.go | 29 +-- executor_test.go | 68 ++++--- field.go | 32 +--- field_internal_test.go | 166 +++++++---------- field_test.go | 57 +++--- fragment.go | 42 +---- fragment_internal_test.go | 373 +++++++++++++++----------------------- holder.go | 59 +++--- holder_internal_test.go | 30 +-- holder_test.go | 35 ++-- http_translator_test.go | 2 +- index.go | 15 +- index_internal_test.go | 11 +- index_test.go | 55 +++--- server.go | 11 +- server/grpc.go | 2 +- server/handler_test.go | 24 +-- server_internal_test.go | 2 +- stats/stats_test.go | 1 - test/cluster.go | 13 +- test/holder.go | 35 ++-- test/index.go | 30 ++- tx_internal_test.go | 6 +- view.go | 23 +-- view_internal_test.go | 40 +--- 28 files changed, 460 insertions(+), 754 deletions(-) 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}