From 66dcaa3ae119c9d088c28b395eb2dd80896a2f76 Mon Sep 17 00:00:00 2001
From: tabokie <xy.tao@outlook.com>
Date: Tue, 1 Aug 2023 14:09:12 +0200
Subject: [PATCH] filter deletion in compaction filter

Signed-off-by: tabokie <xy.tao@outlook.com>
---
 db/compaction/compaction_iterator.cc | 15 ++++++++++-----
 include/rocksdb/compaction_filter.h  | 16 ++++++++++++++++
 2 files changed, 26 insertions(+), 5 deletions(-)

diff --git a/db/compaction/compaction_iterator.cc b/db/compaction/compaction_iterator.cc
index 47f85f48e60..6df4c20901b 100644
--- a/db/compaction/compaction_iterator.cc
+++ b/db/compaction/compaction_iterator.cc
@@ -198,8 +198,10 @@ void CompactionIterator::Next() {
 
 bool CompactionIterator::InvokeFilterIfNeeded(bool* need_skip,
                                               Slice* skip_until) {
+  // Note: Filtering kTypeDeletion is UB. Only `UnsafeFilter` accepts it.
   if (!compaction_filter_ ||
-      (ikey_.type != kTypeValue && ikey_.type != kTypeBlobIndex)) {
+      (ikey_.type != kTypeValue && ikey_.type != kTypeBlobIndex &&
+       ikey_.type != kTypeDeletion)) {
     return true;
   }
   bool error = false;
@@ -210,9 +212,12 @@ bool CompactionIterator::InvokeFilterIfNeeded(bool* need_skip,
   CompactionFilter::Decision filter = CompactionFilter::Decision::kUndetermined;
   compaction_filter_value_.clear();
   compaction_filter_skip_until_.Clear();
-  CompactionFilter::ValueType value_type =
-      ikey_.type == kTypeValue ? CompactionFilter::ValueType::kValue
-                               : CompactionFilter::ValueType::kBlobIndex;
+  CompactionFilter::ValueType value_type = CompactionFilter::ValueType::kValue;
+  if (ikey_.type == kTypeBlobIndex) {
+    value_type = CompactionFilter::ValueType::kBlobIndex;
+  } else if (ikey_.type == kTypeDeletion) {
+    value_type = CompactionFilter::ValueType::kDeletion;
+  }
   // Hack: pass internal key to BlobIndexCompactionFilter since it needs
   // to get sequence number.
   assert(compaction_filter_);
@@ -273,7 +278,7 @@ bool CompactionIterator::InvokeFilterIfNeeded(bool* need_skip,
       }
     }
     if (CompactionFilter::Decision::kUndetermined == filter) {
-      filter = compaction_filter_->FilterV3(
+      filter = compaction_filter_->UnsafeFilter(
           level_, filter_key, ikey_.sequence, value_type,
           blob_value_.empty() ? value_ : blob_value_, &compaction_filter_value_,
           compaction_filter_skip_until_.rep());
diff --git a/include/rocksdb/compaction_filter.h b/include/rocksdb/compaction_filter.h
index f4e4ee13e95..51baba840a3 100644
--- a/include/rocksdb/compaction_filter.h
+++ b/include/rocksdb/compaction_filter.h
@@ -36,6 +36,7 @@ class CompactionFilter : public Customizable {
     kValue,
     kMergeOperand,
     kBlobIndex,  // used internally by BlobDB.
+    kDeletion,   // used only by TiKV's region range filter.
   };
 
   enum class Decision {
@@ -212,6 +213,21 @@ class CompactionFilter : public Customizable {
                     skip_until);
   }
 
+  // This interface is reserved for TiKV's region range filter. Only this
+  // interface can accept `value_type=kTypeDeletion`.
+  virtual Decision UnsafeFilter(int level, const Slice& key,
+                                SequenceNumber seqno, ValueType value_type,
+                                const Slice& existing_value,
+                                std::string* new_value,
+                                std::string* skip_until) const {
+    if (value_type != ValueType::kTypeDeletion) {
+      return FilterV3(level, key, seqno, value_type, existing_value, new_value,
+                      skip_until);
+    } else {
+      return Decision::kKeep;
+    }
+  }
+
   // Internal (BlobDB) use only. Do not override in application code.
   virtual BlobDecision PrepareBlobOutput(const Slice& /* key */,
                                          const Slice& /* existing_value */,