Merge pull request #1816 from molecula/fb-1105

[FB-1105] Fix RBF multi-level branch delete

rbf/cursor.go

@@ -774,6 +774,25 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) {
 	cells[len(cells)-1] = branchCell{}
 	cells = cells[:len(cells)-1]
 
+	// Branches are not allowed to have zero element so we must remove the page
+	// or, in the case of the root page, convert to a leaf page.
+	if len(cells) == 0 {
+		// If this is the root page, convert to leaf page.
+		if stackIndex == 0 {
+			var buf [PageSize]byte
+			writePageNo(buf[:], elem.pgno)
+			writeFlags(buf[:], PageTypeLeaf)
+			writeCellN(buf[:], len(cells))
+			return c.tx.writePage(buf[:])
+		}
+
+		// If this is a non-root page, free and remove from parent.
+		if err := c.tx.freePgno(elem.pgno); err != nil {
+			return err
+		}
+		return c.deleteBranchCell(stackIndex-1, oldPageKey)
+	}
+
 	// If the root only has one node, replace it with its child.
 	if stackIndex == 0 && len(cells) == 1 {
 		target, _, err := c.tx.readPage(cells[0].ChildPgno)
@@ -802,6 +821,9 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) {
 		writeBranchCell(buf[:], j, offset, cell)
 		offset += align8(branchCellSize)
 	}
+
+	assert(readCellN(buf[:]) > 0) // must have at least one cell
+
 	if err := c.tx.writePage(buf[:]); err != nil {
 		return err
 	}

rbf/tx_test.go

@@ -433,6 +433,52 @@ func TestTx_DeallocateToFreeList(t *testing.T) {
 	}
 }
 
+func TestTx_Remove(t *testing.T) {
+	t.Parallel()
+
+	db := MustOpenDB(t)
+	defer MustCloseDB(t, db)
+
+	tx := MustBegin(t, db, true)
+	defer tx.Rollback()
+
+	if err := tx.CreateBitmap("x"); err != nil {
+		t.Fatal(err)
+	}
+
+	// Insert large array values.
+	var values []uint64
+	for i := 0; i < 1000; i++ {
+		for j := 0; j < rbf.ArrayMaxSize; j++ {
+			v := uint64((i << 16) + j)
+			values = append(values, v)
+
+			if _, err := tx.Add("x", v); err != nil {
+				t.Fatalf("Add(%d) err=%q", v, err)
+			}
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+
+	tx = MustBegin(t, db, true)
+	defer tx.Rollback()
+
+	// Remove all array values.
+	for _, i := range rand.Perm(len(values)) {
+		v := values[i]
+		if _, err := tx.Remove("x", v); err != nil {
+			t.Fatalf("Remove(%d) err=%q", v, err)
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestTx_AddRemove_Quick(t *testing.T) {
 	if testing.Short() {
 		t.Skip("-short enabled, skipping")