diff --git a/fragment.go b/fragment.go index 2b2925e16..1718d4acb 100644 --- a/fragment.go +++ b/fragment.go @@ -1733,12 +1733,15 @@ func (f *fragment) bulkImportMutex(rowIDs, columnIDs []uint64) error { // importValue bulk imports a set of range-encoded values. func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear bool) error { + f.mu.Lock() + defer f.mu.Unlock() + // Verify that there are an equal number of column ids and values. if len(columnIDs) != len(values) { return fmt.Errorf("mismatch of column/value len: %d != %d", len(columnIDs), len(values)) } var toSet, toClear []uint64 - f.mu.RLock() + smallWrite := false if len(columnIDs)*int(bitDepth+1)+f.opN < f.MaxOpN { smallWrite = true @@ -1751,14 +1754,11 @@ func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear toSet = make([]uint64, 0, len(columnIDs)*int(bitDepth+1)*(5/8)) toClear = make([]uint64, 0, len(columnIDs)*int(bitDepth+1)*(5/8)) } - f.mu.RUnlock() if !smallWrite { - f.mu.Lock() - defer f.mu.Unlock() - f.storage.OpWriter = nil } + // Process every value. // If an error occurs then reopen the storage. if err := func() (err error) { diff --git a/fragment_internal_test.go b/fragment_internal_test.go index b726d4fe5..71d3b5ebe 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -3024,3 +3024,24 @@ func TestSmallImportRestart(t *testing.T) { t.Errorf("row 1 should be [1], but got %v", r1) } } + +func TestImportValueConcurrent(t *testing.T) { + f := mustOpenFragment("i", "f", viewBSIGroupPrefix+"foo", 0, "none") + eg := &errgroup.Group{} + for i := 0; i < 4; i++ { + i := i + eg.Go(func() error { + for j := uint64(0); j < 10; j++ { + err := f.importValue([]uint64{j}, []uint64{uint64(rand.Int63n(1000))}, 10, i%2 == 0) + if err != nil { + return err + } + } + return nil + }) + } + err := eg.Wait() + if err != nil { + t.Fatalf("concurrently importing values: %v", err) + } +}