Merge pull request #1966 from molecula/fb-1239

[FB-1239] Free previous RBF bitmap pointer page when replacing container

rbf/cursor.go

@@ -287,9 +287,6 @@ func (c *Cursor) Remove(v uint64) (changed bool, err error) {
 
 			leafCell1 := ConvertToLeafArgs(cell.Key, cbm)
 			// ConvertToLeafArgs returns leafCell1 with BitN and ElemN updated.
-			if err := c.tx.freePgno(pgno); err != nil {
-				return false, err
-			}
 			return true, c.putLeafCell(leafCell1)
 		}
 
@@ -375,10 +372,18 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
 		newEstPageSize += in.Size() - len(readLeafCellBytesAtOffset(leafPage, readCellOffset(leafPage, elem.index)))
 	}
 
+	// Use the fast path if we are not splitting pages and the container types are the same.
+	useFast := newEstPageSize+16 <= PageSize
+	if useFast && !isInsert {
+		if prev := readLeafCell(leafPage, elem.index); prev.Type != in.Type {
+			useFast = false
+		}
+	}
+
 	// Use an optimized routine to insert the leaf cell if we won't overflow.
 	// We pad the estimate with 16 bytes because we do 8-byte alignment of
 	// both the cell and the index.
-	if newEstPageSize+16 <= PageSize {
+	if useFast {
 		return c.putLeafCellFast(in, isInsert)
 	}
 
@@ -402,6 +407,22 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
 		copy(cells[elem.index+1:], cells[elem.index:])
 
 	} else {
+		// FB-1239: Free bitmap page if replaced container is a bitmap pointer.
+		prev := cells[elem.index]
+		if prev.Type == ContainerTypeBitmapPtr {
+			if in.Type == ContainerTypeBitmapPtr {
+				if toPgno(in.Data) != toPgno(prev.Data) { // bptr-to-bptr with different bitmap pages
+					if err := c.tx.freePgno(toPgno(prev.Data)); err != nil {
+						return err
+					}
+				}
+			} else if in.Type != ContainerTypeBitmap {
+				if err := c.tx.freePgno(toPgno(prev.Data)); err != nil {
+					return err
+				}
+			}
+		}
+
 		if in.Type == ContainerTypeBitmap {
 			cell = cells[elem.index]
 			if cell.Type != ContainerTypeBitmapPtr {
@@ -433,6 +454,7 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
 		}
 		cell.Data = fromPgno(bitmapPgno)
 	}
+
 	cells[elem.index] = cell
 
 	// Split into multiple pages if page size is exceeded.

rbf/cursor_test.go

@@ -1247,3 +1247,63 @@ func TestForEachRange(t *testing.T) {
 		t.Fatalf("expected empty container, but see %v values left: '%#v'", len(valmap), valmap)
 	}
 }
+
+func TestCursor_PutContainer(t *testing.T) {
+	t.Run("BitmapToArray", func(t *testing.T) {
+		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)
+		}
+
+		bmData := make([]uint64, 1024)
+		for i := range bmData {
+			bmData[i] = 0x5555555555555555
+		}
+		if err := tx.PutContainer("x", 0, roaring.NewContainerBitmap(-1, bmData)); err != nil {
+			t.Fatal(err)
+		}
+		if err := tx.PutContainer("x", 0, roaring.NewContainerArray([]uint16{0})); err != nil {
+			t.Fatal(err)
+		}
+		if err := tx.Commit(); err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	t.Run("BitmapToBitmap", func(t *testing.T) {
+		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)
+		}
+
+		data0 := make([]uint64, 1024)
+		for i := range data0 {
+			data0[i] = 0x5555555555555555
+		}
+		if err := tx.PutContainer("x", 0, roaring.NewContainerBitmap(-1, data0)); err != nil {
+			t.Fatal(err)
+		}
+
+		data1 := make([]uint64, 1024)
+		for i := range data0 {
+			data1[i] = 0x7777777777777777
+		}
+		if err := tx.PutContainer("x", 0, roaring.NewContainerBitmap(-1, data1)); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := tx.Commit(); err != nil {
+			t.Fatal(err)
+		}
+	})
+}