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

adust logic to include normalFlow vs recovery after merge

Browse source
This commit is contained in:
Todd Gruben 2022-02-28 08:12:03 -06:00
parent 0ed85d6d69
commit 376af2c25f
2 changed files with 25 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
executor.go
View file

@ -8292,10 +8292,13 @@ func (e *executor) executeDeleteRecordFromShard(ctx context.Context, qcx *Qcx, i
return
}
return DeleteRows(ctx, src, idx, shard)
return DeleteRowsWithFlow(ctx, src, idx, shard, true)
}
func DeleteRows(ctx context.Context, src *Row, idx *Index, shard uint64) (bool, error) {
return DeleteRowsWithFlow(ctx, src, idx, shard, false)
}
func DeleteRowsWithFlow(ctx context.Context, src *Row, idx *Index, shard uint64, normalFlow bool) (bool, error) {
var existenceFragment *fragment
var deletedRowID uint64
var commitor Commitor = &NopCommitor{}
@ -8310,12 +8313,14 @@ func DeleteRows(ctx context.Context, src *Row, idx *Index, shard uint64) (bool,
if idx.Keys() {
//store columns in exits field ToBeDelete row commited
existenceFragment = idx.Holder().fragment(idx.Name(), existenceFieldName, viewStandard, shard)
if existenceFragment == nil {
//no exists field
return false, errors.New("can't bulk delete without existence field")
if normalFlow { // normalFlow is the standard path, "not normal" is recoverory
existenceFragment = idx.Holder().fragment(idx.Name(), existenceFieldName, viewStandard, shard)
if existenceFragment == nil {
//no exists field
return false, errors.New("can't bulk delete without existence field")
}
deletedRowID, err = transactExistRow(ctx, idx, shard, existenceFragment, src)
}
deletedRowID, err = transactExistRow(ctx, idx, shard, existenceFragment, src)
commitor, err = deleteKeyTranslation(ctx, idx, shard, columns)
if err != nil {
return false, err
@ -8388,7 +8393,18 @@ func DeleteRows(ctx context.Context, src *Row, idx *Index, shard uint64) (bool,
}
close(resChan)
if existenceFragment != nil { //a string keys have been deleted and the deleteRow was created
existenceFragment.clearRow(writeTx, deletedRowID)
if normalFlow {
existenceFragment.clearRow(writeTx, deletedRowID)
} else {
// this is if we are recovering from failure and cleaning up
rows, err := existenceFragment.rows(ctx, writeTx, 1)
if err != nil {
return false, err
}
for _, rowId := range rows {
existenceFragment.clearRow(writeTx, rowId)
}
}
}
return changed, nil
}

5
rbf/cursor.go
View file

@ -474,11 +474,11 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
writeCellN(buf[:], len(group))
offset := dataOffset(len(group))
X := 0
x := 0
for j, cell := range group {
writeLeafCell(buf[:], j, offset, cell)
offset += align8(cell.Size())
X++
x++
}
if err := c.tx.writePage(buf[:]); err != nil {
@ -634,7 +634,6 @@ func (c *Cursor) deleteLeafCell(key uint64) (err error) {
// Update the parent's reference key if it's changed.
if c.stack.top > 0 && oldPageKey != cells[0].Key {
return c.updateBranchCell(c.stack.top-1, cells[0].Key)
}
return nil