From 8f0270acda319a0a252ca7abe1fceda1eb47c72a Mon Sep 17 00:00:00 2001 From: Travis Date: Fri, 12 Feb 2021 18:51:40 -0600 Subject: [PATCH] adjust openExistenceField() to check on disk first --- executor.go | 2 +- field.go | 8 +++-- holder.go | 10 ++++++- holder_test.go | 9 +++++- index.go | 81 ++++++++++++++++++++++++++++++++++++-------------- server.go | 1 - 6 files changed, 81 insertions(+), 30 deletions(-) diff --git a/executor.go b/executor.go index 76974f0e0..93619c432 100644 --- a/executor.go +++ b/executor.go @@ -5067,7 +5067,7 @@ func (e *executor) executeSet(ctx context.Context, qcx *Qcx, index string, c *pq // Set column on existence field. if ef := idx.existenceField(); ef != nil { // we create tx here, rather than just above, to avoid creating an extra empty shard. - tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Field: ef, Shard: shard}) if err != nil { return false, err } diff --git a/field.go b/field.go index 86479f72d..34a6aa72e 100644 --- a/field.go +++ b/field.go @@ -752,7 +752,6 @@ func (f *Field) ForeignIndex() string { // openViews opens and initializes the views inside the field. func (f *Field) openViews() error { - view2shards := f.idx.fieldView2shard.getViewsForField(f.name) if view2shards == nil { // no data @@ -1189,8 +1188,11 @@ func (f *Field) createViewIfNotExistsBase(cvm *CreateViewMessage) (*view, bool, defer f.mu.Unlock() // Create the view in etcd as the system of record. - if err := f.persistView(context.Background(), cvm); err != nil { - return nil, false, errors.Wrap(err, "persisting view") + // Don't persist views related to the existence field. + if f.name != existenceFieldName { + if err := f.persistView(context.Background(), cvm); err != nil { + return nil, false, errors.Wrap(err, "persisting view") + } } if view := f.viewMap[cvm.View]; view != nil { diff --git a/holder.go b/holder.go index 1ee947c4b..91d0e1211 100644 --- a/holder.go +++ b/holder.go @@ -1281,7 +1281,15 @@ func (h *Holder) loadView(indexName, fieldName, viewName string) (*view, error) return nil, errors.Wrap(err, "decoding CreateFieldMessage") } - return fld.createViewIfNotExists(cvm.View) + // I think we eventually want to get rid of storing the serialized view in + // etcd because all it keeps is the view name. So in that case we would always + // just use the viewName argument here. + vName := cvm.View + if fieldName == existenceFieldName { + vName = viewName + } + + return fld.createViewIfNotExists(vName) } func (h *Holder) newIndex(path, name string) (*Index, error) { diff --git a/holder_test.go b/holder_test.go index 376cec189..6904cef88 100644 --- a/holder_test.go +++ b/holder_test.go @@ -32,6 +32,7 @@ import ( func TestHolder_Open(t *testing.T) { t.Run("ErrIndexPermission", func(t *testing.T) { + t.Skip("we don't open the holder directly from disk anymore; we use the etcd schema") if os.Geteuid() == 0 { t.Skip("Skipping permissions test since user is root.") } @@ -50,10 +51,11 @@ func TestHolder_Open(t *testing.T) { }() if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "permission denied") { - t.Fatalf("unexpected error: %s", err) + t.Fatalf("unexpected error: %v", err) } }) t.Run("ErrIndexAttrStoreCorrupt", func(t *testing.T) { + t.Skip("we don't open the holder directly from disk anymore; we use the etcd schema") h := test.MustOpenHolder(t) defer h.Close() @@ -71,6 +73,7 @@ func TestHolder_Open(t *testing.T) { }) t.Run("ErrFieldPermission", func(t *testing.T) { + t.Skip("we don't open the holder directly from disk anymore; we use the etcd schema") if os.Geteuid() == 0 { t.Skip("Skipping permissions test since user is root.") } @@ -94,6 +97,7 @@ func TestHolder_Open(t *testing.T) { } }) t.Run("ErrFieldOptionsCorrupt", func(t *testing.T) { + t.Skip("we don't open the holder directly from disk anymore; we use the etcd schema") h := test.MustOpenHolder(t) defer h.Close() @@ -117,6 +121,7 @@ func TestHolder_Open(t *testing.T) { } }) t.Run("ErrFieldAttrStoreCorrupt", func(t *testing.T) { + t.Skip("we don't open the holder directly from disk anymore; we use the etcd schema") h := test.MustOpenHolder(t) defer h.Close() @@ -140,6 +145,7 @@ func TestHolder_Open(t *testing.T) { }) t.Run("ErrFragmentStoragePermission", func(t *testing.T) { + t.Skip("we don't open the holder directly from disk anymore; we use the etcd schema") roaringOnlyTest(t) if os.Geteuid() == 0 { @@ -177,6 +183,7 @@ func TestHolder_Open(t *testing.T) { } }) t.Run("ErrFragmentStorageCorrupt", func(t *testing.T) { + t.Skip("we don't open the holder directly from disk anymore; we use the etcd schema") roaringOnlyTest(t) h := test.MustOpenHolder(t) diff --git a/index.go b/index.go index b166b9613..ea04dcf45 100644 --- a/index.go +++ b/index.go @@ -330,31 +330,11 @@ fileLoop: }() i.holder.Logger.Debugf("open field: %s", fi.Name()) - mu.Lock() - - // goroutine safe - i.holder.addIndex(i) - - fld, err := i.newField(i.fieldPath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) - fld.createdAt = createdAt - - mu.Unlock() + _, err := i.openField(&mu, createdAt, fi.Name()) if err != nil { - return errors.Wrapf(ErrName, "'%s'", fi.Name()) + return errors.Wrap(err, "opening field") } - // Pass holder through to the field for use in looking - // up a foreign index. - fld.holder = i.holder - - // open the views we have data for. - if err := fld.Open(); err != nil { - return fmt.Errorf("open field: name=%s, err=%s", fld.Name(), err) - } - i.holder.Logger.Debugf("add field to index.fields: %s", fi.Name()) - i.mu.Lock() - i.fields[fld.Name()] = fld - i.mu.Unlock() return nil }) } @@ -371,8 +351,54 @@ fileLoop: return err } +// openField opens the field directory, initializes the field, and adds it to +// the in-memory map of fields maintained by Index. +func (i *Index) openField(mu *sync.Mutex, createdAt int64, file string) (*Field, error) { + mu.Lock() + + // goroutine safe + i.holder.addIndex(i) + + fld, err := i.newField(i.fieldPath(filepath.Base(file)), filepath.Base(file)) + mu.Unlock() + if err != nil { + return nil, errors.Wrapf(ErrName, "'%s'", file) + } + + // Pass holder through to the field for use in looking + // up a foreign index. + fld.holder = i.holder + + fld.createdAt = createdAt + + // open the views we have data for. + if err := fld.Open(); err != nil { + return nil, fmt.Errorf("open field: name=%s, err=%s", fld.Name(), err) + } + + i.holder.Logger.Debugf("add field to index.fields: %s", file) + i.mu.Lock() + i.fields[fld.Name()] = fld + i.mu.Unlock() + + return fld, nil +} + // openExistenceField gets or creates the existence field and associates it to the index. func (i *Index) openExistenceField() error { + // First try opening the existence field from disk. If it doesn't already + // exist on disk, then we fall through to the code path which creates it. + var mu sync.Mutex + fld, err := i.openField(&mu, 0, existenceFieldName) + if err == nil { + i.existenceFld = fld + return nil + } else if errors.Cause(err) != ErrName { + return errors.Wrap(err, "opening existence file") + } + + // If we have gotten here, it means that we couldn't successfully open the + // existence field from disk, so we need to create it. f, err := i.createFieldIfNotExists(existenceFieldName, &FieldOptions{CacheType: CacheTypeNone, CacheSize: 0}) if err != nil { return errors.Wrap(err, "creating existence field") @@ -502,7 +528,9 @@ func (i *Index) Field(name string) *Field { return i.field(name) } -func (i *Index) field(name string) *Field { return i.fields[name] } +func (i *Index) field(name string) *Field { + return i.fields[name] +} // Fields returns a list of all fields in the index. func (i *Index) Fields() []*Field { @@ -692,6 +720,9 @@ func (i *Index) persistField(ctx context.Context, cfm *CreateFieldMessage) error return nil } +// createFieldIfNotExists creates the field if it does not already exist in the +// in-memory index structure. This is not related to whether or not the field +// exists in etcd. func (i *Index) createFieldIfNotExists(name string, opt *FieldOptions) (*Field, error) { i.mu.Lock() defer i.mu.Unlock() @@ -711,6 +742,10 @@ func (i *Index) createFieldIfNotExists(name string, opt *FieldOptions) (*Field, return i.createField(cfm, false) } +// createField, in addition to creating a new Field, calls Field.Open which +// potentially aquires a lock on Index. So until/unless we refactor the +// Index.createField() function call path, we cannot call Index.createField +// while holding an Index lock. func (i *Index) createField(cfm *CreateFieldMessage, broadcast bool) (*Field, error) { opt := cfm.Meta if opt == nil { diff --git a/server.go b/server.go index 2b455085c..2803a2c5d 100644 --- a/server.go +++ b/server.go @@ -557,7 +557,6 @@ func (s *Server) Open() error { if err != nil { return errors.Wrap(err, "starting DisCo") } - _ = initState // Set node ID. s.nodeID = s.disCo.ID()