Repository: cberner/redb
Pull request #80 (2021-10-20 04:20:23)

Fix bug in deletion code path
Deleting a key could delete other keys as well, due to the tree repair
code dropping pages that were not adjacent to the page that needed
repair


Patch:
diff --git a/src/btree.rs b/src/btree.rs
index cf9911c..9bdf064 100644
--- a/src/btree.rs
+++ b/src/btree.rs
@@ -873,6 +873,7 @@ pub(in crate) fn tree_delete<'a, K: RedbKey + ?Sized>(
     (result, freed)
 }
 
+#[derive(Debug)]
 enum DeletionResult {
     // A proper subtree
     Subtree(PageNumber),
@@ -1022,25 +1023,15 @@ fn repair_children(
     } else if children.iter().any(|x| matches!(x, PartialLeaf(_))) {
         let mut result = vec![];
         let mut repaired = false;
+        // For each whole subtree, try to merge it with a partial left to repair it, if one is neighboring
         for i in 0..children.len() {
             if let Subtree(page_number) = &children[i] {
                 if repaired {
                     result.push(*page_number);
                     continue;
                 }
-                if i > 0 {
-                    if let Some(PartialLeaf(partials)) = children.get(i - 1) {
-                        if let Some((page1, page2)) = split_leaf(*page_number, partials, manager) {
-                            result.push(page1);
-                            result.push(page2);
-                            repaired = true;
-                        } else {
-                            result.push(merge_leaf(*page_number, partials, manager));
-                            repaired = true;
-                        }
-                    }
-                }
-                if let Some(PartialLeaf(partials)) = children.get(i + 1) {
+                let offset = if i > 0 { i - 1 } else { i + 1 };
+                if let Some(PartialLeaf(partials)) = children.get(offset) {
                     if let Some((page1, page2)) = split_leaf(*page_number, partials, manager) {
                         result.push(page1);
                         result.push(page2);
@@ -1049,6 +1040,9 @@ fn repair_children(
                         result.push(merge_leaf(*page_number, partials, manager));
                         repaired = true;
                     }
+                } else {
+                    // No adjacent partial
+                    result.push(*page_number);
                 }
             }
         }
@@ -1056,25 +1050,15 @@ fn repair_children(
     } else if children.iter().any(|x| matches!(x, PartialInternal(_))) {
         let mut result = vec![];
         let mut repaired = false;
+        // For each whole subtree, try to merge it with a partial left to repair it, if one is neighboring
         for i in 0..children.len() {
             if let Subtree(page_number) = &children[i] {
                 if repaired {
                     result.push(*page_number);
                     continue;
                 }
-                if i > 0 {
-                    if let Some(PartialInternal(partials)) = children.get(i - 1) {
-                        if let Some((page1, page2)) = split_index(*page_number, partials, manager) {
-                            result.push(page1);
-                            result.push(page2);
-                            repaired = true;
-                        } else {
-                            result.push(merge_index(*page_number, partials, manager));
-                            repaired = true;
-                        }
-                    }
-                }
-                if let Some(PartialInternal(partials)) = children.get(i + 1) {
+                let offset = if i > 0 { i - 1 } else { i + 1 };
+                if let Some(PartialInternal(partials)) = children.get(offset) {
                     if let Some((page1, page2)) = split_index(*page_number, partials, manager) {
                         result.push(page1);
                         result.push(page2);
@@ -1083,6 +1067,9 @@ fn repair_children(
                         result.push(merge_index(*page_number, partials, manager));
                         repaired = true;
                     }
+                } else {
+                    // No adjacent partial
+                    result.push(*page_number);
                 }
             }
         }


Test patch:
diff --git a/tests/integration_tests.rs b/tests/integration_tests.rs
index 6c9e918..7e25da1 100644
--- a/tests/integration_tests.rs
+++ b/tests/integration_tests.rs
@@ -120,3 +120,45 @@ fn large_keys() {
     }
     txn.commit().unwrap();
 }
+
+#[test]
+// Test for a bug in the deletion code, where deleting a key accidentally deleted other keys
+fn regression() {
+    let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
+
+    let db_size = 1024 * 1024;
+    let db = unsafe { Database::open(tmpfile.path(), db_size).unwrap() };
+    let mut table: Table<u64, u64> = db.open_table(b"x").unwrap();
+
+    let mut txn = table.begin_write().unwrap();
+    txn.insert(&1, &1).unwrap();
+    txn.commit().unwrap();
+
+    let mut txn = table.begin_write().unwrap();
+    txn.insert(&6, &9).unwrap();
+    txn.commit().unwrap();
+
+    let mut txn = table.begin_write().unwrap();
+    txn.insert(&12, &10).unwrap();
+    txn.commit().unwrap();
+
+    let mut txn = table.begin_write().unwrap();
+    txn.insert(&18, &27).unwrap();
+    txn.commit().unwrap();
+
+    let mut txn = table.begin_write().unwrap();
+    txn.insert(&24, &33).unwrap();
+    txn.commit().unwrap();
+
+    let mut txn = table.begin_write().unwrap();
+    txn.insert(&30, &14).unwrap();
+    txn.commit().unwrap();
+
+    let mut txn = table.begin_write().unwrap();
+    txn.remove(&30).unwrap();
+    txn.commit().unwrap();
+
+    let txn = table.read_transaction().unwrap();
+    let v = txn.get(&6).unwrap().unwrap().to_value();
+    assert_eq!(v, 9);
+}