fix: deletion between rows with the same key may not work

GreptimeTeam · Dec 14, 2024 · 863f474 · 863f474
1 parent 579059d
commit 863f474
Showing 1 changed file with 32 additions and 0 deletions.
diff --git a/src/mito2/src/read/dedup.rs b/src/mito2/src/read/dedup.rs
@@ -311,6 +311,12 @@ impl LastFieldsBuilder {
             return;
         }
 
+        self.contains_deletion = batch.op_types().get_data(0).unwrap() == OpType::Delete as u8;
+        if self.contains_deletion {
+            // Deletes this row.
+            return;
+        }
+
         let fields = batch.fields();
         for (idx, value) in self.last_fields.iter_mut().enumerate() {
             if value.is_null() && !fields[idx].data.is_null(0) {
@@ -1082,6 +1088,32 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_last_non_null_strategy_delete_middle() {
+        let input = [
+            new_batch_multi_fields(b"k1", &[1], &[7], &[OpType::Put], &[(Some(11), None)]),
+            new_batch_multi_fields(b"k1", &[1], &[4], &[OpType::Delete], &[(None, None)]),
+            new_batch_multi_fields(b"k1", &[1], &[1], &[OpType::Put], &[(Some(12), Some(1))]),
+            new_batch_multi_fields(b"k1", &[2], &[8], &[OpType::Put], &[(Some(21), None)]),
+            new_batch_multi_fields(b"k1", &[2], &[5], &[OpType::Delete], &[(None, None)]),
+            new_batch_multi_fields(b"k1", &[2], &[2], &[OpType::Put], &[(Some(22), Some(2))]),
+            new_batch_multi_fields(b"k1", &[3], &[9], &[OpType::Put], &[(Some(31), None)]),
+            new_batch_multi_fields(b"k1", &[3], &[6], &[OpType::Delete], &[(None, None)]),
+            new_batch_multi_fields(b"k1", &[3], &[3], &[OpType::Put], &[(Some(32), Some(3))]),
+        ];
+
+        let mut strategy = LastNonNull::new(true);
+        check_dedup_strategy(
+            &input,
+            &mut strategy,
+            &[
+                new_batch_multi_fields(b"k1", &[1], &[7], &[OpType::Put], &[(Some(11), None)]),
+                new_batch_multi_fields(b"k1", &[2], &[8], &[OpType::Put], &[(Some(21), None)]),
+                new_batch_multi_fields(b"k1", &[3], &[9], &[OpType::Put], &[(Some(31), None)]),
+            ],
+        );
+    }
+
     #[test]
     fn test_last_non_null_iter_on_batch() {
         let input = [new_batch_multi_fields(