scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-10 23:11:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #2003 from molecula/fb1306

Browse source 
[FB-1306] ingest API delete could deadlock
This commit is contained in:
seebs 2022-04-04 11:38:08 -05:00 • committed by GitHub
commit e379ae5a36
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 24 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
api.go
View file

@ -1919,7 +1919,7 @@ func (api *API) applyOperations(ctx context.Context, qcx *Qcx, index *Index, sha
// We delete from the existence field unconditionally and other fields
// if we know they exist.
if op.OpType == ingest.OpDelete {
err = clearExistenceColumns(qcx, index, op.ClearRecordIDs, shard)
err = clearExistenceColumns(tx, index, op.ClearRecordIDs, shard)
if err != nil {
return fmt.Errorf("clearing existence columns: %w", err)
}
@ -1997,21 +1997,20 @@ func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard ui
return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, &options)
}
func clearExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uint64) error {
func clearExistenceColumns(tx Tx, index *Index, columnIDs []uint64, shard uint64) error {
ef := index.existenceField()
if ef == nil {
return nil
}
existenceRowIDs := make([]uint64, len(columnIDs))
// If we don't gratuitously hand-duplicate things in field.Import,
// the fact that fragment.bulkImport rewrites its row and column
// lists can burn us if we don't make a copy before doing the
// existence field write.
columnCopy := make([]uint64, len(columnIDs))
copy(columnCopy, columnIDs)
options := ImportOptions{Clear: true}
return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, &options)
v := ef.view("standard")
if v == nil {
return nil
}
f := v.Fragment(shard)
if f == nil {
return nil
}
return f.ClearRecords(tx, columnIDs)
}
// ShardDistribution returns an object representing the distribution of shards

3
ingest_test.go
View file

@ -146,6 +146,9 @@ func parseExpectedResults(data []byte) (ints []uint64, keys []string, err error)
return nil, nil, errors.New("expecting [] results")
}
words := bytes.Split(data[1:len(data)-1], []byte{','})
if len(words) == 1 && len(words[0]) == 0 {
return nil, nil, nil
}
for _, word := range words {
word = bytes.TrimSpace(word)
if len(word) == 0 {

10
ingest_testdata/sample.tc
View file

@ -87,3 +87,13 @@ ingest-error:
}
}
]
ingest:
[
{
"action": "delete",
"record_ids": [ 1 ]
}
]
queries:
Row(setkey="a")
[]